Mirror
/
zulip
mirror of https://github.com/zulip/zulip.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

logging: Migrate many backend tests to use assertLogs. 

This commit  migrates some of the backend tests to use assertLogs(),
instead of mock.patch() as planned in #15331.

Tweaked by tabbott to avoid tautological assertions.
This commit is contained in:
m-e-l-u-h-a-n 2020-12-24 02:15:16 +05:30 committed by Tim Abbott
parent 7417ac9165
commit ccf520ff13
9 changed files with 133 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
corporate/tests/test_stripe.py
View File

@ -359,20 +359,16 @@ class StripeTestCase(ZulipTestCase):
upgrade_func(*args) upgrade_func(*args)
class StripeTest(StripeTestCase): class StripeTest(StripeTestCase):
@patch("corporate.lib.stripe.billing_logger.info") def test_catch_stripe_errors(self) -> None:
@patch("corporate.lib.stripe.billing_logger.error")
def test_catch_stripe_errors(self, mock_billing_logger_error: Mock, mock_billing_logger_info: Mock) -> None:
@catch_stripe_errors @catch_stripe_errors
def raise_invalid_request_error() -> None: def raise_invalid_request_error() -> None:
raise stripe.error.InvalidRequestError( raise stripe.error.InvalidRequestError(
"message", "param", "code", json_body={}) "message", "param", "code", json_body={})
with self.assertRaises(BillingError) as context: with self.assertLogs('corporate.stripe', "ERROR") as error_log:
raise_invalid_request_error() with self.assertRaises(BillingError) as context:
self.assertEqual('other stripe error', context.exception.description) raise_invalid_request_error()
mock_billing_logger_error.assert_called_once() self.assertEqual('other stripe error', context.exception.description)
mock_billing_logger_info.assert_not_called() self.assertEqual(error_log.output, ['ERROR:corporate.stripe:Stripe error: None None None None'])
mock_billing_logger_error.reset_mock()
mock_billing_logger_info.reset_mock()
@catch_stripe_errors @catch_stripe_errors
def raise_card_error() -> None: def raise_card_error() -> None:
@ -380,12 +376,12 @@ class StripeTest(StripeTestCase):
json_body = {"error": {"message": error_message}} json_body = {"error": {"message": error_message}}
raise stripe.error.CardError(error_message, "number", "invalid_number", raise stripe.error.CardError(error_message, "number", "invalid_number",
json_body=json_body) json_body=json_body)
with self.assertRaises(StripeCardError) as context: with self.assertLogs('corporate.stripe', "INFO") as info_log:
raise_card_error() with self.assertRaises(StripeCardError) as context:
self.assertIn('not a valid credit card', context.exception.message) raise_card_error()
self.assertEqual('card error', context.exception.description) self.assertIn('not a valid credit card', context.exception.message)
mock_billing_logger_info.assert_called_once() self.assertEqual('card error', context.exception.description)
mock_billing_logger_error.assert_not_called() self.assertEqual(info_log.output, ['INFO:corporate.stripe:Stripe card error: None None None None'])
def test_billing_not_enabled(self) -> None: def test_billing_not_enabled(self) -> None:
iago = self.example_user('iago') iago = self.example_user('iago')
@ -901,9 +897,9 @@ class StripeTest(StripeTestCase):
self.login_user(user) self.login_user(user)
# From https://stripe.com/docs/testing#cards: Attaching this card to # From https://stripe.com/docs/testing#cards: Attaching this card to
# a Customer object succeeds, but attempts to charge the customer fail. # a Customer object succeeds, but attempts to charge the customer fail.
with patch("corporate.lib.stripe.billing_logger.info") as mock_billing_logger: with self.assertLogs('corporate.stripe', 'INFO') as m:
self.upgrade(stripe_token=stripe_create_token('4000000000000341').id) self.upgrade(stripe_token=stripe_create_token('4000000000000341').id)
mock_billing_logger.assert_called_once() self.assertEqual(m.output, ['INFO:corporate.stripe:Stripe card error: 402 card_error card_declined None'])
# Check that we created a Customer object but no CustomerPlan # Check that we created a Customer object but no CustomerPlan
stripe_customer_id = Customer.objects.get(realm=get_realm('zulip')).stripe_customer_id stripe_customer_id = Customer.objects.get(realm=get_realm('zulip')).stripe_customer_id
self.assertFalse(CustomerPlan.objects.exists()) self.assertFalse(CustomerPlan.objects.exists())
@ -973,11 +969,12 @@ class StripeTest(StripeTestCase):
hamlet = self.example_user('hamlet') hamlet = self.example_user('hamlet')
self.login_user(hamlet) self.login_user(hamlet)
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token') self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token')
with patch("corporate.lib.stripe.billing_logger.warning") as mock_billing_logger: with self.assertLogs('corporate.stripe', "WARNING") as m:
with self.assertRaises(BillingError) as context: with self.assertRaises(BillingError) as context:
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token') self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token')
self.assertEqual('subscribing with existing subscription', context.exception.description) self.assertEqual('subscribing with existing subscription', context.exception.description)
mock_billing_logger.assert_called_once() self.assertRegexpMatches(m.output[0], r'WARNING:corporate.stripe:Customer <Customer <Realm: zulip \d*> id> trying to upgrade, but has an active subscription')
self.assertEqual(len(m.output), 1)
def test_check_upgrade_parameters(self) -> None: def test_check_upgrade_parameters(self) -> None:
# Tests all the error paths except 'not enough licenses' # Tests all the error paths except 'not enough licenses'
@ -1055,12 +1052,14 @@ class StripeTest(StripeTestCase):
# Invoice # Invoice
check_success(True, MAX_INVOICED_LICENSES) check_success(True, MAX_INVOICED_LICENSES)
@patch("corporate.lib.stripe.billing_logger.error") def test_upgrade_with_uncaught_exception(self) -> None:
def test_upgrade_with_uncaught_exception(self, mock_: Mock) -> None:
hamlet = self.example_user('hamlet') hamlet = self.example_user('hamlet')
self.login_user(hamlet) self.login_user(hamlet)
with patch("corporate.views.process_initial_upgrade", side_effect=Exception): with patch("corporate.views.process_initial_upgrade", side_effect=Exception), \
self.assertLogs('corporate.stripe', "WARNING") as m:
response = self.upgrade(talk_to_stripe=False) response = self.upgrade(talk_to_stripe=False)
self.assertIn('ERROR:corporate.stripe:Uncaught exception in billing', m.output[0])
self.assertIn(m.records[0].stack_info, m.output[0])
self.assert_json_error_contains(response, "Something went wrong. Please contact desdemona+admin@zulip.com.") self.assert_json_error_contains(response, "Something went wrong. Please contact desdemona+admin@zulip.com.")
self.assertEqual(orjson.loads(response.content)['error_description'], 'uncaught exception during upgrade') self.assertEqual(orjson.loads(response.content)['error_description'], 'uncaught exception during upgrade')
@ -1298,11 +1297,10 @@ class StripeTest(StripeTestCase):
# Replace with an invalid card # Replace with an invalid card
stripe_token = stripe_create_token(card_number='4000000000009987').id stripe_token = stripe_create_token(card_number='4000000000009987').id
with patch("corporate.lib.stripe.billing_logger.info") as mock_billing_logger: with patch("stripe.Invoice.list") as mock_invoice_list, self.assertLogs('corporate.stripe', 'INFO') as m:
with patch("stripe.Invoice.list") as mock_invoice_list: response = self.client_post("/json/billing/sources/change",
response = self.client_post("/json/billing/sources/change", {'stripe_token': orjson.dumps(stripe_token).decode()})
{'stripe_token': orjson.dumps(stripe_token).decode()}) self.assertEqual(m.output, ['INFO:corporate.stripe:Stripe card error: 402 card_error card_declined '])
mock_billing_logger.assert_called_once()
mock_invoice_list.assert_not_called() mock_invoice_list.assert_not_called()
self.assertEqual(orjson.loads(response.content)['error_description'], 'card error') self.assertEqual(orjson.loads(response.content)['error_description'], 'card error')
self.assert_json_error_contains(response, 'Your card was declined') self.assert_json_error_contains(response, 'Your card was declined')
@ -1313,10 +1311,10 @@ class StripeTest(StripeTestCase):
# Replace with a card that's valid, but charging the card fails # Replace with a card that's valid, but charging the card fails
stripe_token = stripe_create_token(card_number='4000000000000341').id stripe_token = stripe_create_token(card_number='4000000000000341').id
with patch("corporate.lib.stripe.billing_logger.info") as mock_billing_logger: with self.assertLogs('corporate.stripe', 'INFO') as m:
response = self.client_post("/json/billing/sources/change", response = self.client_post("/json/billing/sources/change",
{'stripe_token': orjson.dumps(stripe_token).decode()}) {'stripe_token': orjson.dumps(stripe_token).decode()})
mock_billing_logger.assert_called_once() self.assertEqual(m.output, ['INFO:corporate.stripe:Stripe card error: 402 card_error card_declined None'])
self.assertEqual(orjson.loads(response.content)['error_description'], 'card error') self.assertEqual(orjson.loads(response.content)['error_description'], 'card error')
self.assert_json_error_contains(response, 'Your card was declined') self.assert_json_error_contains(response, 'Your card was declined')
for stripe_source in stripe_get_customer(stripe_customer_id).sources: for stripe_source in stripe_get_customer(stripe_customer_id).sources:
@ -1344,14 +1342,15 @@ class StripeTest(StripeTestCase):
self.assertEqual(2, RealmAuditLog.objects.filter( self.assertEqual(2, RealmAuditLog.objects.filter(
event_type=RealmAuditLog.STRIPE_CARD_CHANGED).count()) event_type=RealmAuditLog.STRIPE_CARD_CHANGED).count())
@patch("corporate.lib.stripe.billing_logger.info") def test_downgrade(self) -> None:
def test_downgrade(self, mock_: Mock) -> None:
user = self.example_user("hamlet") user = self.example_user("hamlet")
self.login_user(user) self.login_user(user)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now): with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token') self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token')
response = self.client_post("/json/billing/plan/change", with self.assertLogs('corporate.stripe', 'INFO') as m:
{'status': CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE}) response = self.client_post("/json/billing/plan/change",
{'status': CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE})
self.assertRegexpMatches(m.output[0], r"INFO:corporate.stripe:Change plan status: Customer.id: \d*, CustomerPlan.id: \d*, status: \d*")
self.assert_json_success(response) self.assert_json_success(response)
# Verify that we still write LicenseLedger rows during the remaining # Verify that we still write LicenseLedger rows during the remaining
@ -1402,7 +1401,6 @@ class StripeTest(StripeTestCase):
mocked.assert_not_called() mocked.assert_not_called()
@mock_stripe() @mock_stripe()
@patch("corporate.lib.stripe.billing_logger.info")
def test_switch_from_monthly_plan_to_annual_plan_for_automatic_license_management(self, *mocks: Mock) -> None: def test_switch_from_monthly_plan_to_annual_plan_for_automatic_license_management(self, *mocks: Mock) -> None:
user = self.example_user("hamlet") user = self.example_user("hamlet")
@ -1414,8 +1412,10 @@ class StripeTest(StripeTestCase):
self.assertEqual(monthly_plan.automanage_licenses, True) self.assertEqual(monthly_plan.automanage_licenses, True)
self.assertEqual(monthly_plan.billing_schedule, CustomerPlan.MONTHLY) self.assertEqual(monthly_plan.billing_schedule, CustomerPlan.MONTHLY)
response = self.client_post("/json/billing/plan/change", with self.assertLogs('corporate.stripe', "INFO") as m:
{'status': CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE}) response = self.client_post("/json/billing/plan/change",
{'status': CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE})
self.assertRegexpMatches(m.output[0], r'INFO:corporate.stripe:Change plan status: Customer.id: \d*, CustomerPlan.id: \d*, status: 4')
self.assert_json_success(response) self.assert_json_success(response)
monthly_plan.refresh_from_db() monthly_plan.refresh_from_db()
self.assertEqual(monthly_plan.status, CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE) self.assertEqual(monthly_plan.status, CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE)
@ -1545,7 +1545,6 @@ class StripeTest(StripeTestCase):
self.assertEqual(invoice_item[key], value) self.assertEqual(invoice_item[key], value)
@mock_stripe() @mock_stripe()
@patch("corporate.lib.stripe.billing_logger.info")
def test_switch_from_monthly_plan_to_annual_plan_for_manual_license_management(self, *mocks: Mock) -> None: def test_switch_from_monthly_plan_to_annual_plan_for_manual_license_management(self, *mocks: Mock) -> None:
user = self.example_user("hamlet") user = self.example_user("hamlet")
num_licenses = 35 num_licenses = 35
@ -1557,9 +1556,10 @@ class StripeTest(StripeTestCase):
assert(monthly_plan is not None) assert(monthly_plan is not None)
self.assertEqual(monthly_plan.automanage_licenses, False) self.assertEqual(monthly_plan.automanage_licenses, False)
self.assertEqual(monthly_plan.billing_schedule, CustomerPlan.MONTHLY) self.assertEqual(monthly_plan.billing_schedule, CustomerPlan.MONTHLY)
with self.assertLogs('corporate.stripe', "INFO") as m:
response = self.client_post("/json/billing/plan/change", response = self.client_post("/json/billing/plan/change",
{'status': CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE}) {'status': CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE})
self.assertRegexpMatches(m.output[0], r'INFO:corporate.stripe:Change plan status: Customer.id: \d*, CustomerPlan.id: \d*, status: 4')
self.assert_json_success(response) self.assert_json_success(response)
monthly_plan.refresh_from_db() monthly_plan.refresh_from_db()
self.assertEqual(monthly_plan.status, CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE) self.assertEqual(monthly_plan.status, CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE)
@ -1633,23 +1633,24 @@ class StripeTest(StripeTestCase):
for key, value in annual_plan_invoice_item_params.items(): for key, value in annual_plan_invoice_item_params.items():
self.assertEqual(invoice_item[key], value) self.assertEqual(invoice_item[key], value)
@patch("corporate.lib.stripe.billing_logger.info") def test_reupgrade_after_plan_status_changed_to_downgrade_at_end_of_cycle(self) -> None:
def test_reupgrade_after_plan_status_changed_to_downgrade_at_end_of_cycle(self, mock_: Mock) -> None:
user = self.example_user("hamlet") user = self.example_user("hamlet")
self.login_user(user) self.login_user(user)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now): with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token') self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token')
response = self.client_post("/json/billing/plan/change", with self.assertLogs('corporate.stripe', 'INFO') as m:
{'status': CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE}) response = self.client_post("/json/billing/plan/change",
{'status': CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE})
self.assertRegexpMatches(m.output[0], r'INFO:corporate.stripe:Change plan status: Customer.id: \d*, CustomerPlan.id: \d*, status: 2')
self.assert_json_success(response) self.assert_json_success(response)
self.assertEqual(CustomerPlan.objects.first().status, CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE) self.assertEqual(CustomerPlan.objects.first().status, CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE)
with self.assertLogs('corporate.stripe', "INFO") as m:
response = self.client_post("/json/billing/plan/change", response = self.client_post("/json/billing/plan/change",
{'status': CustomerPlan.ACTIVE}) {'status': CustomerPlan.ACTIVE})
self.assertRegexpMatches(m.output[0], r'INFO:corporate.stripe:Change plan status: Customer.id: \d*, CustomerPlan.id: \d*, status: 1')
self.assert_json_success(response) self.assert_json_success(response)
self.assertEqual(CustomerPlan.objects.first().status, CustomerPlan.ACTIVE) self.assertEqual(CustomerPlan.objects.first().status, CustomerPlan.ACTIVE)
@patch("corporate.lib.stripe.billing_logger.info")
@patch("stripe.Invoice.create") @patch("stripe.Invoice.create")
@patch("stripe.Invoice.finalize_invoice") @patch("stripe.Invoice.finalize_invoice")
@patch("stripe.InvoiceItem.create") @patch("stripe.InvoiceItem.create")
@ -1663,8 +1664,10 @@ class StripeTest(StripeTestCase):
self.login_user(user) self.login_user(user)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now): with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token') self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token')
self.client_post("/json/billing/plan/change", with self.assertLogs('corporate.stripe', "INFO") as m:
{'status': CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE}) self.client_post("/json/billing/plan/change",
{'status': CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE})
self.assertRegexpMatches(m.output[0], r'INFO:corporate.stripe:Change plan status: Customer.id: \d*, CustomerPlan.id: \d*, status: 2')
plan = CustomerPlan.objects.first() plan = CustomerPlan.objects.first()
self.assertIsNotNone(plan.next_invoice_date) self.assertIsNotNone(plan.next_invoice_date)
@ -1718,8 +1721,6 @@ class StripeTest(StripeTestCase):
invoice_plans_as_needed(self.next_year) invoice_plans_as_needed(self.next_year)
mocked.assert_not_called() mocked.assert_not_called()
@patch("corporate.lib.stripe.billing_logger.warning")
@patch("corporate.lib.stripe.billing_logger.info")
def test_reupgrade_by_billing_admin_after_downgrade(self, *mocks: Mock) -> None: def test_reupgrade_by_billing_admin_after_downgrade(self, *mocks: Mock) -> None:
user = self.example_user("hamlet") user = self.example_user("hamlet")
@ -1727,12 +1728,15 @@ class StripeTest(StripeTestCase):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token') self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token')
self.login_user(user) self.login_user(user)
self.client_post("/json/billing/plan/change", with self.assertLogs('corporate.stripe', 'INFO') as m:
{'status': CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE}) self.client_post("/json/billing/plan/change",
{'status': CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE})
self.assertRegexpMatches(m.output[0], r'INFO:corporate.stripe:Change plan status: Customer.id: \d*, CustomerPlan.id: \d*, status: 2')
with self.assertRaises(BillingError) as context: with self.assertRaises(BillingError) as context, self.assertLogs('corporate.stripe', 'WARNING') as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now): with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token') self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, 'token')
self.assertRegexpMatches(m.output[0], r'WARNING:corporate.stripe:Customer <Customer <Realm: zulip \d*> id> trying to upgrade, but has an active subscription')
self.assertEqual(context.exception.description, "subscribing with existing subscription") self.assertEqual(context.exception.description, "subscribing with existing subscription")
invoice_plans_as_needed(self.next_year) invoice_plans_as_needed(self.next_year)

55
zerver/tests/test_cache.py
View File

@ -29,12 +29,13 @@ from zerver.models import UserProfile, get_system_bot, get_user_profile_by_email
class AppsTest(ZulipTestCase): class AppsTest(ZulipTestCase):
def test_cache_gets_flushed(self) -> None: def test_cache_gets_flushed(self) -> None:
with patch('zerver.apps.logging.info') as mock_logging: with self.assertLogs(level="INFO") as m:
with patch('zerver.apps.cache.clear') as mock: with patch('zerver.apps.cache.clear') as mock:
# The argument to flush_cache doesn't matter # The argument to flush_cache doesn't matter
flush_cache(Mock()) flush_cache(Mock())
mock.assert_called_once() mock.assert_called_once()
mock_logging.assert_called_once() self.assertEqual(m.output, ['INFO:root:Clearing memcached cache after migrations'])
self.assertEqual(len(m.output), 1)
class CacheKeyValidationTest(ZulipTestCase): class CacheKeyValidationTest(ZulipTestCase):
def test_validate_cache_key(self) -> None: def test_validate_cache_key(self) -> None:
@ -85,14 +86,14 @@ class CacheWithKeyDecoratorTest(ZulipTestCase):
hamlet = self.example_user('hamlet') hamlet = self.example_user('hamlet')
with patch('zerver.lib.cache.cache_set') as mock_set, \ with patch('zerver.lib.cache.cache_set') as mock_set, \
patch('zerver.lib.cache.logger.warning') as mock_warn: self.assertLogs(level="WARNING") as m:
with queries_captured() as queries: with queries_captured() as queries:
result = get_user_function_with_bad_cache_keys(hamlet.id) result = get_user_function_with_bad_cache_keys(hamlet.id)
self.assertEqual(len(m.output), 1)
self.assertEqual(result, hamlet) self.assertEqual(result, hamlet)
self.assert_length(queries, 1) self.assert_length(queries, 1)
mock_set.assert_not_called() mock_set.assert_not_called()
mock_warn.assert_called_once()
def test_cache_with_key_key_too_long(self) -> None: def test_cache_with_key_key_too_long(self) -> None:
def too_long_cache_key_function(user_id: int) -> str: def too_long_cache_key_function(user_id: int) -> str:
@ -105,14 +106,14 @@ class CacheWithKeyDecoratorTest(ZulipTestCase):
hamlet = self.example_user('hamlet') hamlet = self.example_user('hamlet')
with patch('zerver.lib.cache.cache_set') as mock_set, \ with patch('zerver.lib.cache.cache_set') as mock_set, \
patch('zerver.lib.cache.logger.warning') as mock_warn: self.assertLogs(level="WARNING") as m:
with queries_captured() as queries: with queries_captured() as queries:
result = get_user_function_with_bad_cache_keys(hamlet.id) result = get_user_function_with_bad_cache_keys(hamlet.id)
self.assertEqual(len(m.output), 1)
self.assertEqual(result, hamlet) self.assertEqual(result, hamlet)
self.assert_length(queries, 1) self.assert_length(queries, 1)
mock_set.assert_not_called() mock_set.assert_not_called()
mock_warn.assert_called_once()
def test_cache_with_key_good_key(self) -> None: def test_cache_with_key_good_key(self) -> None:
def good_cache_key_function(user_id: int) -> str: def good_cache_key_function(user_id: int) -> str:
@ -193,10 +194,10 @@ class GetCacheWithKeyDecoratorTest(ZulipTestCase):
return return
hamlet = self.example_user('hamlet') hamlet = self.example_user('hamlet')
with patch('zerver.lib.cache.logger.warning') as mock_warn: with self.assertLogs(level='WARNING') as m:
with self.assertRaises(NotFoundInCache): with self.assertRaises(NotFoundInCache):
get_user_function_with_bad_cache_keys(hamlet.id) get_user_function_with_bad_cache_keys(hamlet.id)
mock_warn.assert_called_once() self.assertEqual(len(m.output), 1)
class SafeCacheFunctionsTest(ZulipTestCase): class SafeCacheFunctionsTest(ZulipTestCase):
def test_safe_cache_functions_with_all_good_keys(self) -> None: def test_safe_cache_functions_with_all_good_keys(self) -> None:
@ -209,46 +210,32 @@ class SafeCacheFunctionsTest(ZulipTestCase):
def test_safe_cache_functions_with_all_bad_keys(self) -> None: def test_safe_cache_functions_with_all_bad_keys(self) -> None:
items = {"SafeFunctionsTest:\nbadkey1": 1, "SafeFunctionsTest:\nbadkey2": 2} items = {"SafeFunctionsTest:\nbadkey1": 1, "SafeFunctionsTest:\nbadkey2": 2}
with patch('zerver.lib.cache.logger.warning') as mock_warn: with self.assertLogs(level="WARNING") as m:
safe_cache_set_many(items) safe_cache_set_many(items)
mock_warn.assert_called_once() self.assertIn('WARNING:root:Invalid cache key used: [\'SafeFunctionsTest:\\nbadkey1\', \'SafeFunctionsTest:\\nbadkey2\']', m.output[0])
self.assertEqual( self.assertEqual(len(m.output), 1)
mock_warn.call_args[0][1],
['SafeFunctionsTest:\nbadkey1', 'SafeFunctionsTest:\nbadkey2'],
)
with patch('zerver.lib.cache.logger.warning') as mock_warn: with self.assertLogs(level="WARNING") as m:
result = safe_cache_get_many(list(items.keys())) result = safe_cache_get_many(list(items.keys()))
mock_warn.assert_called_once()
self.assertEqual(
mock_warn.call_args[0][1],
['SafeFunctionsTest:\nbadkey1', 'SafeFunctionsTest:\nbadkey2'],
)
self.assertEqual(result, {}) self.assertEqual(result, {})
self.assertIn('WARNING:root:Invalid cache key used: [\'SafeFunctionsTest:\\nbadkey1\', \'SafeFunctionsTest:\\nbadkey2\']', m.output[0])
self.assertEqual(len(m.output), 1)
def test_safe_cache_functions_with_good_and_bad_keys(self) -> None: def test_safe_cache_functions_with_good_and_bad_keys(self) -> None:
bad_items = {"SafeFunctionsTest:\nbadkey1": 1, "SafeFunctionsTest:\nbadkey2": 2} bad_items = {"SafeFunctionsTest:\nbadkey1": 1, "SafeFunctionsTest:\nbadkey2": 2}
good_items = {"SafeFunctionsTest:goodkey1": 3, "SafeFunctionsTest:goodkey2": 4} good_items = {"SafeFunctionsTest:goodkey1": 3, "SafeFunctionsTest:goodkey2": 4}
items = {**good_items, **bad_items} items = {**good_items, **bad_items}
with patch('zerver.lib.cache.logger.warning') as mock_warn: with self.assertLogs(level="WARNING") as m:
safe_cache_set_many(items) safe_cache_set_many(items)
mock_warn.assert_called_once() self.assertIn('WARNING:root:Invalid cache key used: [\'SafeFunctionsTest:\\nbadkey1\', \'SafeFunctionsTest:\\nbadkey2\']', m.output[0])
self.assertEqual( self.assertEqual(len(m.output), 1)
mock_warn.call_args[0][1],
['SafeFunctionsTest:\nbadkey1', 'SafeFunctionsTest:\nbadkey2'],
)
with patch('zerver.lib.cache.logger.warning') as mock_warn: with self.assertLogs(level="WARNING") as m:
result = safe_cache_get_many(list(items.keys())) result = safe_cache_get_many(list(items.keys()))
mock_warn.assert_called_once()
self.assertEqual(
mock_warn.call_args[0][1],
['SafeFunctionsTest:\nbadkey1', 'SafeFunctionsTest:\nbadkey2'],
)
self.assertEqual(result, good_items) self.assertEqual(result, good_items)
self.assertIn('WARNING:root:Invalid cache key used: [\'SafeFunctionsTest:\\nbadkey1\', \'SafeFunctionsTest:\\nbadkey2\']', m.output[0])
self.assertEqual(len(m.output), 1)
class BotCacheKeyTest(ZulipTestCase): class BotCacheKeyTest(ZulipTestCase):
def test_bot_profile_key_deleted_on_save(self) -> None: def test_bot_profile_key_deleted_on_save(self) -> None:

11
zerver/tests/test_embedded_bot_system.py
View File

@ -66,11 +66,11 @@ class TestEmbeddedBotMessaging(ZulipTestCase):
assert self.bot_profile is not None assert self.bot_profile is not None
with patch('zulip_bots.bots.helloworld.helloworld.HelloWorldHandler.handle_message', with patch('zulip_bots.bots.helloworld.helloworld.HelloWorldHandler.handle_message',
side_effect=EmbeddedBotQuitException("I'm quitting!")): side_effect=EmbeddedBotQuitException("I'm quitting!")):
with patch('logging.warning') as mock_logging: with self.assertLogs(level='WARNING') as m:
self.send_stream_message(self.user_profile, "Denmark", self.send_stream_message(self.user_profile, "Denmark",
content=f"@**{self.bot_profile.full_name}** foo", content=f"@**{self.bot_profile.full_name}** foo",
topic_name="bar") topic_name="bar")
mock_logging.assert_called_once_with("I'm quitting!") self.assertEqual(m.output, ['WARNING:root:I\'m quitting!'])
class TestEmbeddedBotFailures(ZulipTestCase): class TestEmbeddedBotFailures(ZulipTestCase):
def test_message_embedded_bot_with_invalid_service(self) -> None: def test_message_embedded_bot_with_invalid_service(self) -> None:
@ -83,11 +83,8 @@ class TestEmbeddedBotFailures(ZulipTestCase):
service_profile = get_service_profile(bot_profile.id, 'helloworld') service_profile = get_service_profile(bot_profile.id, 'helloworld')
service_profile.name = 'invalid' service_profile.name = 'invalid'
service_profile.save() service_profile.save()
with patch('logging.error') as logging_error_mock: with self.assertLogs(level='ERROR') as m:
self.send_stream_message(user_profile, "Denmark", self.send_stream_message(user_profile, "Denmark",
content=f"@**{bot_profile.full_name}** foo", content=f"@**{bot_profile.full_name}** foo",
topic_name="bar") topic_name="bar")
logging_error_mock.assert_called_once_with( self.assertRegexpMatches(m.output[0], r'ERROR:root:Error: User [0-9]* has bot with invalid embedded bot service invalid')
"Error: User %s has bot with invalid embedded bot service %s",
bot_profile.id, "invalid",
)

5
zerver/tests/test_home.py
View File

@ -455,10 +455,9 @@ class HomeTest(ZulipTestCase):
def test_bad_narrow(self) -> None: def test_bad_narrow(self) -> None:
self.login('hamlet') self.login('hamlet')
with patch('logging.warning') as mock: with self.assertLogs(level='WARNING') as m:
result = self._get_home_page(stream='Invalid Stream') result = self._get_home_page(stream='Invalid Stream')
mock.assert_called_once() self.assertEqual(m.output, ['WARNING:root:Invalid narrow requested, ignoring'])
self.assertEqual(mock.call_args_list[0][0][0], "Invalid narrow requested, ignoring")
self._sanity_check(result) self._sanity_check(result)
def test_topic_narrow(self) -> None: def test_topic_narrow(self) -> None:

12
zerver/tests/test_queue.py
View File

@ -10,13 +10,13 @@ from zerver.lib.test_classes import ZulipTestCase
class TestTornadoQueueClient(ZulipTestCase): class TestTornadoQueueClient(ZulipTestCase):
@mock.patch('zerver.lib.queue.logging.getLogger', autospec=True)
@mock.patch('zerver.lib.queue.ExceptionFreeTornadoConnection', autospec=True) @mock.patch('zerver.lib.queue.ExceptionFreeTornadoConnection', autospec=True)
def test_on_open_closed(self, mock_cxn: mock.MagicMock, def test_on_open_closed(self, mock_cxn: mock.MagicMock) -> None:
mock_get_logger: mock.MagicMock) -> None: with self.assertLogs('zulip.queue', 'WARNING') as m:
connection = TornadoQueueClient() connection = TornadoQueueClient()
connection.connection.channel.side_effect = ConnectionClosed('500', 'test') connection.connection.channel.side_effect = ConnectionClosed('500', 'test')
connection._on_open(mock.MagicMock()) connection._on_open(mock.MagicMock())
self.assertEqual(m.output, ['WARNING:zulip.queue:TornadoQueueClient couldn\'t open channel: connection already closed'])
class TestQueueImplementation(ZulipTestCase): class TestQueueImplementation(ZulipTestCase):

44
zerver/tests/test_queue_worker.py
View File

@ -27,7 +27,6 @@ from zerver.worker.queue_processors import (
LoopQueueProcessingWorker, LoopQueueProcessingWorker,
MissedMessageWorker, MissedMessageWorker,
QueueProcessingWorker, QueueProcessingWorker,
WorkerTimeoutException,
get_active_worker_queues, get_active_worker_queues,
) )
@ -303,11 +302,9 @@ class WorkerTest(ZulipTestCase):
self.assertEqual(mock_mirror_email.call_count, 3) self.assertEqual(mock_mirror_email.call_count, 3)
@patch('zerver.lib.rate_limiter.logger.warning')
@patch('zerver.worker.queue_processors.mirror_email') @patch('zerver.worker.queue_processors.mirror_email')
@override_settings(RATE_LIMITING_MIRROR_REALM_RULES=[(10, 2)]) @override_settings(RATE_LIMITING_MIRROR_REALM_RULES=[(10, 2)])
def test_mirror_worker_rate_limiting(self, mock_mirror_email: MagicMock, def test_mirror_worker_rate_limiting(self, mock_mirror_email: MagicMock) -> None:
mock_warn: MagicMock) -> None:
fake_client = self.FakeClient() fake_client = self.FakeClient()
realm = get_realm('zulip') realm = get_realm('zulip')
RateLimitedRealmMirror(realm).clear_history() RateLimitedRealmMirror(realm).clear_history()
@ -360,13 +357,11 @@ class WorkerTest(ZulipTestCase):
# If RateLimiterLockingException is thrown, we rate-limit the new message: # If RateLimiterLockingException is thrown, we rate-limit the new message:
with patch('zerver.lib.rate_limiter.RedisRateLimiterBackend.incr_ratelimit', with patch('zerver.lib.rate_limiter.RedisRateLimiterBackend.incr_ratelimit',
side_effect=RateLimiterLockingException): side_effect=RateLimiterLockingException):
fake_client.enqueue('email_mirror', data[0]) with self.assertLogs('zerver.lib.rate_limiter', 'WARNING') as mock_warn:
worker.start() fake_client.enqueue('email_mirror', data[0])
self.assertEqual(mock_mirror_email.call_count, 4) worker.start()
mock_warn.assert_called_with( self.assertEqual(mock_mirror_email.call_count, 4)
"Deadlock trying to incr_ratelimit for %s", self.assertEqual(mock_warn.output, ['WARNING:zerver.lib.rate_limiter:Deadlock trying to incr_ratelimit for RateLimitedRealmMirror:zulip'])
f"RateLimitedRealmMirror:{realm.string_id}",
)
self.assertEqual(warn_logs.output, [ self.assertEqual(warn_logs.output, [
'WARNING:zerver.worker.queue_processors:MirrorWorker: Rejecting an email from: None to realm: Zulip Dev - rate limited.' 'WARNING:zerver.worker.queue_processors:MirrorWorker: Rejecting an email from: None to realm: Zulip Dev - rate limited.'
] * 5) ] * 5)
@ -397,8 +392,9 @@ class WorkerTest(ZulipTestCase):
side_effect=smtplib.SMTPServerDisconnected), \ side_effect=smtplib.SMTPServerDisconnected), \
mock_queue_publish('zerver.lib.queue.queue_json_publish', mock_queue_publish('zerver.lib.queue.queue_json_publish',
side_effect=fake_publish), \ side_effect=fake_publish), \
patch('logging.exception'): self.assertLogs(level="ERROR") as m:
worker.start() worker.start()
self.assertIn("failed due to exception SMTPServerDisconnected", m.output[0])
self.assertEqual(data['failed_tries'], 1 + MAX_REQUEST_RETRIES) self.assertEqual(data['failed_tries'], 1 + MAX_REQUEST_RETRIES)
@ -451,12 +447,10 @@ class WorkerTest(ZulipTestCase):
with simulated_queue_client(lambda: fake_client): with simulated_queue_client(lambda: fake_client):
worker = UnreliableWorker() worker = UnreliableWorker()
worker.setup() worker.setup()
with patch('logging.exception') as logging_exception_mock: with self.assertLogs(level='ERROR') as m:
worker.start() worker.start()
logging_exception_mock.assert_called_once_with( self.assertEqual(m.records[0].message, 'Problem handling data on queue unreliable_worker')
"Problem handling data on queue %s", "unreliable_worker", self.assertIn(m.records[0].stack_info, m.output[0])
stack_info=True,
)
self.assertEqual(processed, ['good', 'fine', 'back to normal']) self.assertEqual(processed, ['good', 'fine', 'back to normal'])
with open(fn) as f: with open(fn) as f:
@ -488,12 +482,10 @@ class WorkerTest(ZulipTestCase):
with simulated_queue_client(lambda: fake_client): with simulated_queue_client(lambda: fake_client):
loopworker = UnreliableLoopWorker() loopworker = UnreliableLoopWorker()
loopworker.setup() loopworker.setup()
with patch('logging.exception') as logging_exception_mock: with self.assertLogs(level='ERROR') as m:
loopworker.start() loopworker.start()
logging_exception_mock.assert_called_once_with( self.assertEqual(m.records[0].message, 'Problem handling data on queue unreliable_loopworker')
"Problem handling data on queue %s", "unreliable_loopworker", self.assertIn(m.records[0].stack_info, m.output[0])
stack_info=True,
)
self.assertEqual(processed, ['good', 'fine']) self.assertEqual(processed, ['good', 'fine'])
with open(fn) as f: with open(fn) as f:
@ -530,12 +522,10 @@ class WorkerTest(ZulipTestCase):
worker = TimeoutWorker() worker = TimeoutWorker()
worker.setup() worker.setup()
worker.ENABLE_TIMEOUTS = True worker.ENABLE_TIMEOUTS = True
with patch('logging.exception') as logging_exception_mock: with self.assertLogs(level='ERROR') as m:
worker.start() worker.start()
logging_exception_mock.assert_called_once_with( self.assertEqual(m.records[0].message, 'Timed out after 1 seconds processing 1 events in queue timeout_worker')
"%s in queue %s", str(WorkerTimeoutException(1, 1)), "timeout_worker", self.assertIn(m.records[0].stack_info, m.output[0])
stack_info=True,
)
self.assertEqual(processed, ['good', 'fine', 'back to normal']) self.assertEqual(processed, ['good', 'fine', 'back to normal'])
with open(fn) as f: with open(fn) as f:

4
zerver/tests/test_sessions.py
View File

@ -124,7 +124,7 @@ class TestExpirableSessionVars(ZulipTestCase):
def test_get_var_is_not_expirable(self) -> None: def test_get_var_is_not_expirable(self) -> None:
self.session["test_get_var_is_not_expirable"] = 0 self.session["test_get_var_is_not_expirable"] = 0
with mock.patch('zerver.lib.sessions.logging.warning') as mock_warn: with self.assertLogs(level='WARNING') as m:
value = get_expirable_session_var(self.session, 'test_get_var_is_not_expirable', default_value='default') value = get_expirable_session_var(self.session, 'test_get_var_is_not_expirable', default_value='default')
self.assertEqual(value, 'default') self.assertEqual(value, 'default')
mock_warn.assert_called_once() self.assertIn('WARNING:root:get_expirable_session_var: error getting test_get_var_is_not_expirable', m.output[0])

55
zerver/tests/test_signup.py
View File

@ -426,9 +426,9 @@ class PasswordResetTest(ZulipTestCase):
do_deactivate_realm(user_profile.realm) do_deactivate_realm(user_profile.realm)
# start the password reset process by supplying an email address # start the password reset process by supplying an email address
with patch('logging.info') as mock_logging: with self.assertLogs(level="INFO") as m:
result = self.client_post('/accounts/password/reset/', {'email': email}) result = self.client_post('/accounts/password/reset/', {'email': email})
mock_logging.assert_called_once() self.assertEqual(m.output, ["INFO:root:Realm is deactivated"])
# check the redirect link telling you to check mail for password reset link # check the redirect link telling you to check mail for password reset link
self.assertEqual(result.status_code, 302) self.assertEqual(result.status_code, 302)
@ -523,9 +523,9 @@ class PasswordResetTest(ZulipTestCase):
def test_ldap_auth_only(self) -> None: def test_ldap_auth_only(self) -> None:
"""If the email auth backend is not enabled, password reset should do nothing""" """If the email auth backend is not enabled, password reset should do nothing"""
email = self.example_email("hamlet") email = self.example_email("hamlet")
with patch('logging.info') as mock_logging: with self.assertLogs(level="INFO") as m:
result = self.client_post('/accounts/password/reset/', {'email': email}) result = self.client_post('/accounts/password/reset/', {'email': email})
mock_logging.assert_called_once() self.assertEqual(m.output, ["INFO:root:Password reset attempted for hamlet@zulip.com even though password auth is disabled."])
# check the redirect link telling you to check mail for password reset link # check the redirect link telling you to check mail for password reset link
self.assertEqual(result.status_code, 302) self.assertEqual(result.status_code, 302)
@ -547,9 +547,9 @@ class PasswordResetTest(ZulipTestCase):
# If the domain matches, we don't generate an email # If the domain matches, we don't generate an email
with self.settings(LDAP_APPEND_DOMAIN="zulip.com"): with self.settings(LDAP_APPEND_DOMAIN="zulip.com"):
email = self.example_email("hamlet") email = self.example_email("hamlet")
with patch('logging.info') as mock_logging: with self.assertLogs(level="INFO") as m:
result = self.client_post('/accounts/password/reset/', {'email': email}) result = self.client_post('/accounts/password/reset/', {'email': email})
mock_logging.assert_called_once_with("Password reset not allowed for user in LDAP domain") self.assertEqual(m.output, ["INFO:root:Password reset not allowed for user in LDAP domain"])
from django.core.mail import outbox from django.core.mail import outbox
self.assertEqual(len(outbox), 0) self.assertEqual(len(outbox), 0)
@ -649,9 +649,9 @@ class LoginTest(ZulipTestCase):
self.assert_logged_in_user_id(None) self.assert_logged_in_user_id(None)
def test_login_wrong_subdomain(self) -> None: def test_login_wrong_subdomain(self) -> None:
with patch("logging.warning") as mock_warning: with self.assertLogs(level="WARNING") as m:
result = self.login_with_return(self.mit_email("sipbtest"), "xxx") result = self.login_with_return(self.mit_email("sipbtest"), "xxx")
mock_warning.assert_called_once() self.assertEqual(m.output, ["WARNING:root:User sipbtest@mit.edu attempted password login to wrong subdomain zulip"])
self.assertEqual(result.status_code, 200) self.assertEqual(result.status_code, 200)
self.assert_in_response("Your Zulip account is not a member of the " self.assert_in_response("Your Zulip account is not a member of the "
"organization associated with this subdomain.", result) "organization associated with this subdomain.", result)
@ -2686,17 +2686,12 @@ class UserSignUpTest(InviteUserBase):
side_effect=smtplib.SMTPException('uh oh'), side_effect=smtplib.SMTPException('uh oh'),
) )
error_mock = patch('logging.error') with smtp_mock, self.assertLogs(level="ERROR") as m:
with smtp_mock, error_mock as err:
result = self.client_post('/accounts/home/', {'email': email}) result = self.client_post('/accounts/home/', {'email': email})
self._assert_redirected_to(result, '/config-error/smtp') self._assert_redirected_to(result, '/config-error/smtp')
self.assertEqual( self.assertEqual(m.output, ['ERROR:root:Error in accounts_home: uh oh'])
err.call_args_list[0][0],
('Error in accounts_home: %s', 'uh oh'),
)
def test_bad_email_configuration_for_create_realm(self) -> None: def test_bad_email_configuration_for_create_realm(self) -> None:
""" """
@ -2709,17 +2704,11 @@ class UserSignUpTest(InviteUserBase):
side_effect=smtplib.SMTPException('uh oh'), side_effect=smtplib.SMTPException('uh oh'),
) )
error_mock = patch('logging.error') with smtp_mock, self.assertLogs(level="ERROR") as m:
with smtp_mock, error_mock as err:
result = self.client_post('/new/', {'email': email}) result = self.client_post('/new/', {'email': email})
self._assert_redirected_to(result, '/config-error/smtp') self._assert_redirected_to(result, '/config-error/smtp')
self.assertEqual(m.output, ['ERROR:root:Error in create_realm: uh oh'])
self.assertEqual(
err.call_args_list[0][0],
('Error in create_realm: %s', 'uh oh'),
)
def test_user_default_language_and_timezone(self) -> None: def test_user_default_language_and_timezone(self) -> None:
""" """
@ -3225,14 +3214,14 @@ class UserSignUpTest(InviteUserBase):
return_data['invalid_subdomain'] = True return_data['invalid_subdomain'] = True
with patch('zerver.views.registration.authenticate', side_effect=invalid_subdomain): with patch('zerver.views.registration.authenticate', side_effect=invalid_subdomain):
with patch('logging.error') as mock_error: with self.assertLogs(level="ERROR") as m:
result = self.client_post( result = self.client_post(
'/accounts/register/', '/accounts/register/',
{'password': password, {'password': password,
'full_name': 'New User', 'full_name': 'New User',
'key': find_key_by_email(email), 'key': find_key_by_email(email),
'terms': True}) 'terms': True})
mock_error.assert_called_once() self.assertEqual(m.output, ["ERROR:root:Subdomain mismatch in registration zulip: newuser@zulip.com"])
self.assertEqual(result.status_code, 302) self.assertEqual(result.status_code, 302)
def test_replace_subdomain_in_confirmation_link(self) -> None: def test_replace_subdomain_in_confirmation_link(self) -> None:
@ -3722,18 +3711,15 @@ class UserSignUpTest(InviteUserBase):
LDAP_APPEND_DOMAIN='example.com', LDAP_APPEND_DOMAIN='example.com',
AUTH_LDAP_USER_ATTR_MAP=ldap_user_attr_map, AUTH_LDAP_USER_ATTR_MAP=ldap_user_attr_map,
): ):
with patch('zerver.views.registration.logging.warning') as mock_warning: with self.assertLogs(level="WARNING") as m:
result = self.submit_reg_form_for_user( result = self.submit_reg_form_for_user(
email, email,
password, password,
from_confirmation="1", from_confirmation="1",
# Pass HTTP_HOST for the target subdomain # Pass HTTP_HOST for the target subdomain
HTTP_HOST=subdomain + ".testserver") HTTP_HOST=subdomain + ".testserver")
self.assertEqual(result.status_code, 200) self.assertEqual(result.status_code, 200)
mock_warning.assert_called_once_with( self.assertEqual(m.output, ['WARNING:root:New account email newuser@zulip.com could not be found in LDAP'])
"New account email %s could not be found in LDAP",
"newuser@zulip.com",
)
with self.assertLogs('zulip.ldap', 'DEBUG') as debug_log: with self.assertLogs('zulip.ldap', 'DEBUG') as debug_log:
result = self.submit_reg_form_for_user(email, result = self.submit_reg_form_for_user(email,
password, password,
@ -3814,7 +3800,7 @@ class UserSignUpTest(InviteUserBase):
LDAP_EMAIL_ATTR='mail', LDAP_EMAIL_ATTR='mail',
AUTH_LDAP_USER_ATTR_MAP=ldap_user_attr_map, AUTH_LDAP_USER_ATTR_MAP=ldap_user_attr_map,
): ):
with patch('zerver.views.registration.logging.warning') as mock_warning: with self.assertLogs(level="WARNING") as m:
result = self.submit_reg_form_for_user( result = self.submit_reg_form_for_user(
email, email,
password, password,
@ -3822,10 +3808,7 @@ class UserSignUpTest(InviteUserBase):
# Pass HTTP_HOST for the target subdomain # Pass HTTP_HOST for the target subdomain
HTTP_HOST=subdomain + ".testserver") HTTP_HOST=subdomain + ".testserver")
self.assertEqual(result.status_code, 200) self.assertEqual(result.status_code, 200)
mock_warning.assert_called_once_with( self.assertEqual(m.output, ['WARNING:root:New account email nonexistent@zulip.com could not be found in LDAP'])
"New account email %s could not be found in LDAP",
"nonexistent@zulip.com",
)
with self.assertLogs('zulip.ldap', 'DEBUG') as debug_log: with self.assertLogs('zulip.ldap', 'DEBUG') as debug_log:
result = self.submit_reg_form_for_user(email, result = self.submit_reg_form_for_user(email,

7
zerver/tests/test_zephyr.py
View File

@ -41,9 +41,6 @@ class ZephyrTest(ZulipTestCase):
def mirror_mock() -> Any: def mirror_mock() -> Any:
return self.settings(PERSONAL_ZMIRROR_SERVER='server') return self.settings(PERSONAL_ZMIRROR_SERVER='server')
def logging_mock() -> Any:
return patch('logging.exception')
cred = dict(cname=dict(nameString=['starnine'])) cred = dict(cname=dict(nameString=['starnine']))
with ccache_mock(side_effect=KeyError('foo')): with ccache_mock(side_effect=KeyError('foo')):
@ -53,11 +50,11 @@ class ZephyrTest(ZulipTestCase):
with \ with \
ccache_mock(return_value=b'1234'), \ ccache_mock(return_value=b'1234'), \
ssh_mock(side_effect=subprocess.CalledProcessError(1, [])), \ ssh_mock(side_effect=subprocess.CalledProcessError(1, [])), \
logging_mock() as log: self.assertLogs(level='ERROR') as log:
result = post("zephyr", cred=cred) result = post("zephyr", cred=cred)
self.assert_json_error(result, 'We were unable to setup mirroring for you') self.assert_json_error(result, 'We were unable to setup mirroring for you')
log.assert_called_with("Error updating the user's ccache", stack_info=True) self.assertIn("Error updating the user's ccache", log.output[0])
with ccache_mock(return_value=b'1234'), mirror_mock(), ssh_mock() as ssh: with ccache_mock(return_value=b'1234'), mirror_mock(), ssh_mock() as ssh:
result = post("zephyr", cred=cred) result = post("zephyr", cred=cred)