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

python: Switch mocking of timezone_now to time_machine.

This commit is contained in:
Alex Vandiver 2023-11-28 18:16:58 +00:00 committed by Tim Abbott
parent 82c08dd153
commit 61fc838179
16 changed files with 189 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
analytics/tests/test_counts.py
View File

@ -130,7 +130,7 @@ class AnalyticsTestCase(ZulipTestCase):
for key, value in defaults.items():
kwargs[key] = kwargs.get(key, value)
kwargs["delivery_email"] = kwargs["email"]
with mock.patch("zerver.lib.create_user.timezone_now", return_value=kwargs["date_joined"]):
with time_machine.travel(kwargs["date_joined"], tick=False):
pass_kwargs: Dict[str, Any] = {}
if kwargs["is_bot"]:
pass_kwargs["bot_type"] = UserProfile.DEFAULT_BOT

90
analytics/tests/test_support_views.py
View File

@ -1,8 +1,9 @@
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Optional
from typing import TYPE_CHECKING, Any, Optional
from unittest import mock
import orjson
import time_machine
from django.utils.timezone import now as timezone_now
from typing_extensions import override
@ -356,55 +357,56 @@ class TestSupportEndpoint(ZulipTestCase):
check_zulip_realm_query_result(result)
check_lear_realm_query_result(result)
with mock.patch(
"analytics.views.support.timezone_now",
return_value=timezone_now() - timedelta(minutes=50),
):
self.client_post("/accounts/home/", {"email": self.nonreg_email("test")})
self.login("iago")
result = get_check_query_result(self.nonreg_email("test"), 1)
check_preregistration_user_query_result(result, self.nonreg_email("test"))
check_zulip_realm_query_result(result)
self.client_post("/accounts/home/", {"email": self.nonreg_email("test")})
self.login("iago")
create_invitation("Denmark", self.nonreg_email("test1"))
result = get_check_query_result(self.nonreg_email("test1"), 1)
check_preregistration_user_query_result(result, self.nonreg_email("test1"), invite=True)
check_zulip_realm_query_result(result)
def query_result_from_before(*args: Any) -> "TestHttpResponse":
with time_machine.travel((timezone_now() - timedelta(minutes=50)), tick=False):
return get_check_query_result(*args)
email = self.nonreg_email("alice")
self.submit_realm_creation_form(
email, realm_subdomain="custom-test", realm_name="Zulip test"
)
result = get_check_query_result(email, 1)
check_realm_creation_query_result(result, email)
result = query_result_from_before(self.nonreg_email("test"), 1)
check_preregistration_user_query_result(result, self.nonreg_email("test"))
check_zulip_realm_query_result(result)
invite_expires_in_minutes = 10 * 24 * 60
do_create_multiuse_invite_link(
self.example_user("hamlet"),
invited_as=1,
invite_expires_in_minutes=invite_expires_in_minutes,
)
result = get_check_query_result("zulip", 2)
check_multiuse_invite_link_query_result(result)
check_zulip_realm_query_result(result)
MultiuseInvite.objects.all().delete()
create_invitation("Denmark", self.nonreg_email("test1"))
result = query_result_from_before(self.nonreg_email("test1"), 1)
check_preregistration_user_query_result(result, self.nonreg_email("test1"), invite=True)
check_zulip_realm_query_result(result)
do_send_realm_reactivation_email(get_realm("zulip"), acting_user=None)
result = get_check_query_result("zulip", 2)
check_realm_reactivation_link_query_result(result)
check_zulip_realm_query_result(result)
email = self.nonreg_email("alice")
self.submit_realm_creation_form(
email, realm_subdomain="custom-test", realm_name="Zulip test"
)
result = query_result_from_before(email, 1)
check_realm_creation_query_result(result, email)
lear_nonreg_email = "newguy@lear.org"
self.client_post("/accounts/home/", {"email": lear_nonreg_email}, subdomain="lear")
result = get_check_query_result(lear_nonreg_email, 1)
check_preregistration_user_query_result(result, lear_nonreg_email)
check_lear_realm_query_result(result)
invite_expires_in_minutes = 10 * 24 * 60
do_create_multiuse_invite_link(
self.example_user("hamlet"),
invited_as=1,
invite_expires_in_minutes=invite_expires_in_minutes,
)
result = query_result_from_before("zulip", 2)
check_multiuse_invite_link_query_result(result)
check_zulip_realm_query_result(result)
MultiuseInvite.objects.all().delete()
self.login_user(lear_user)
create_invitation("general", "newguy2@lear.org", lear_realm)
result = get_check_query_result("newguy2@lear.org", 1, lear_realm.string_id)
check_preregistration_user_query_result(result, "newguy2@lear.org", invite=True)
check_lear_realm_query_result(result)
do_send_realm_reactivation_email(get_realm("zulip"), acting_user=None)
result = query_result_from_before("zulip", 2)
check_realm_reactivation_link_query_result(result)
check_zulip_realm_query_result(result)
lear_nonreg_email = "newguy@lear.org"
self.client_post("/accounts/home/", {"email": lear_nonreg_email}, subdomain="lear")
result = query_result_from_before(lear_nonreg_email, 1)
check_preregistration_user_query_result(result, lear_nonreg_email)
check_lear_realm_query_result(result)
self.login_user(lear_user)
create_invitation("general", "newguy2@lear.org", lear_realm)
result = query_result_from_before("newguy2@lear.org", 1, lear_realm.string_id)
check_preregistration_user_query_result(result, "newguy2@lear.org", invite=True)
check_lear_realm_query_result(result)
def test_get_org_type_display_name(self) -> None:
self.assertEqual(get_org_type_display_name(Realm.ORG_TYPES["business"]["id"]), "Business")

158
corporate/tests/test_stripe.py
View File

@ -28,6 +28,7 @@ import orjson
import responses
import stripe
import stripe.util
import time_machine
from django.conf import settings
from django.core import signing
from django.urls.resolvers import get_resolver
@ -626,7 +627,7 @@ class StripeTestCase(ZulipTestCase):
def add_card_and_upgrade(self, user: UserProfile, **kwargs: Any) -> stripe.Customer:
# Add card
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.add_card_to_customer_for_upgrade()
# Check that we correctly created a Customer object in Stripe
@ -635,7 +636,7 @@ class StripeTestCase(ZulipTestCase):
)
self.assertTrue(stripe_customer_has_credit_card_as_default_payment_method(stripe_customer))
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.upgrade(**kwargs)
self.assert_json_success(response)
@ -868,7 +869,7 @@ class StripeTest(StripeTestCase):
self.assertEqual("/billing/", response["Location"])
# Check /billing/ has the correct information
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_get("/billing/")
self.assert_not_in_success_response(["Pay annually"], response)
for substring in [
@ -910,7 +911,7 @@ class StripeTest(StripeTestCase):
user = self.example_user("hamlet")
self.login_user(user)
# Click "Make payment" in Stripe Checkout
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.upgrade(invoice=True)
# Check that we correctly created a Customer in Stripe
stripe_customer = stripe_get_customer(
@ -1016,7 +1017,7 @@ class StripeTest(StripeTestCase):
self.assertEqual("/billing/", response["Location"])
# Check /billing/ has the correct information
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_get("/billing/")
self.assert_not_in_success_response(["Pay annually", "Update card"], response)
for substring in [
@ -1046,7 +1047,7 @@ class StripeTest(StripeTestCase):
self.assertFalse(Customer.objects.filter(realm=user.realm).exists())
# Require free trial users to add a credit card.
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
with self.assertLogs("corporate.stripe", "WARNING"):
response = self.upgrade()
self.assert_json_error(
@ -1126,7 +1127,7 @@ class StripeTest(StripeTestCase):
self.assertEqual(realm.plan_type, Realm.PLAN_TYPE_STANDARD)
self.assertEqual(realm.max_invites, Realm.INVITES_STANDARD_REALM_DAILY_MAX)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_get("/billing/")
self.assert_not_in_success_response(["Pay annually"], response)
for substring in [
@ -1242,7 +1243,7 @@ class StripeTest(StripeTestCase):
plan.fixed_price = 127
plan.price_per_license = None
plan.save(update_fields=["fixed_price", "price_per_license"])
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_get("/billing/")
self.assert_in_success_response(["$1.27"], response)
# Don't show price breakdown
@ -1263,7 +1264,7 @@ class StripeTest(StripeTestCase):
self.assertNotEqual(user.realm.plan_type, Realm.PLAN_TYPE_STANDARD)
self.assertFalse(Customer.objects.filter(realm=user.realm).exists())
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.upgrade(invoice=True)
stripe_customer = stripe_get_customer(
@ -1333,7 +1334,7 @@ class StripeTest(StripeTestCase):
self.assertEqual(realm.plan_type, Realm.PLAN_TYPE_STANDARD)
self.assertEqual(realm.max_invites, Realm.INVITES_STANDARD_REALM_DAILY_MAX)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_get("/billing/")
self.assert_not_in_success_response(["Pay annually"], response)
for substring in [
@ -2202,14 +2203,14 @@ class StripeTest(StripeTestCase):
def test_downgrade(self) -> None:
user = self.example_user("hamlet")
self.login_user(user)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
plan = get_current_plan_by_realm(user.realm)
assert plan is not None
self.assertEqual(plan.licenses(), self.seat_count)
self.assertEqual(plan.licenses_at_next_renewal(), self.seat_count)
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_patch(
"/json/billing/plan", {"status": CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE}
)
@ -2223,7 +2224,7 @@ class StripeTest(StripeTestCase):
self.assertEqual(plan.licenses(), self.seat_count)
self.assertEqual(plan.licenses_at_next_renewal(), None)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
mock_customer = Mock(email=user.delivery_email)
mock_customer.invoice_settings.default_payment_method = Mock(
spec=stripe.PaymentMethod, type=Mock()
@ -2327,7 +2328,7 @@ class StripeTest(StripeTestCase):
assert new_plan is not None
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_patch(
"/json/billing/plan",
{"status": CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE},
@ -2337,7 +2338,7 @@ class StripeTest(StripeTestCase):
self.assert_json_success(response)
monthly_plan.refresh_from_db()
self.assertEqual(monthly_plan.status, CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_get("/billing/")
self.assert_in_success_response(
["Your plan will switch to annual billing on February 2, 2012"], response
@ -2353,7 +2354,7 @@ class StripeTest(StripeTestCase):
(20, 20),
)
with patch("corporate.lib.stripe.timezone_now", return_value=self.next_month):
with time_machine.travel(self.next_month, tick=False):
with patch("corporate.lib.stripe.get_latest_seat_count", return_value=25):
update_license_ledger_if_needed(user.realm, self.next_month)
self.assertEqual(LicenseLedger.objects.filter(plan=monthly_plan).count(), 2)
@ -2516,7 +2517,7 @@ class StripeTest(StripeTestCase):
new_plan = get_current_plan_by_realm(user.realm)
assert new_plan is not None
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_patch(
"/json/billing/plan",
{"status": CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE},
@ -2528,7 +2529,7 @@ class StripeTest(StripeTestCase):
self.assert_json_success(response)
monthly_plan.refresh_from_db()
self.assertEqual(monthly_plan.status, CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_get("/billing/")
self.assert_in_success_response(
["Your plan will switch to annual billing on February 2, 2012"], response
@ -2629,8 +2630,9 @@ class StripeTest(StripeTestCase):
new_plan = get_current_plan_by_realm(user.realm)
assert new_plan is not None
assert self.now is not None
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_patch(
"/json/billing/plan",
{"status": CustomerPlan.SWITCH_TO_MONTHLY_AT_END_OF_CYCLE},
@ -2640,7 +2642,7 @@ class StripeTest(StripeTestCase):
self.assert_json_success(response)
annual_plan.refresh_from_db()
self.assertEqual(annual_plan.status, CustomerPlan.SWITCH_TO_MONTHLY_AT_END_OF_CYCLE)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_get("/billing/")
self.assert_in_success_response(
["Your plan will switch to monthly billing on January 2, 2013"], response
@ -2659,9 +2661,9 @@ class StripeTest(StripeTestCase):
# Check that we don't switch to monthly plan at next invoice date (which is used to charge user for
# additional licenses) but at the end of current billing cycle.
self.assertEqual(annual_plan.next_invoice_date, self.next_month)
with patch("corporate.lib.stripe.timezone_now", return_value=annual_plan.next_invoice_date):
assert annual_plan.next_invoice_date is not None
with time_machine.travel(annual_plan.next_invoice_date, tick=False):
with patch("corporate.lib.stripe.get_latest_seat_count", return_value=25):
assert annual_plan.next_invoice_date is not None
update_license_ledger_if_needed(user.realm, annual_plan.next_invoice_date)
annual_plan.refresh_from_db()
@ -2715,7 +2717,7 @@ class StripeTest(StripeTestCase):
self.assertEqual(invoice_item2[key], value)
# Check that we switch to monthly plan at the end of current billing cycle.
with patch("corporate.lib.stripe.timezone_now", return_value=self.next_year):
with time_machine.travel(self.next_year, tick=False):
with patch("corporate.lib.stripe.get_latest_seat_count", return_value=25):
update_license_ledger_if_needed(user.realm, self.next_year)
self.assertEqual(LicenseLedger.objects.filter(plan=annual_plan).count(), 3)
@ -2783,7 +2785,7 @@ class StripeTest(StripeTestCase):
for key, value in monthly_plan_invoice_item_params.items():
self.assertEqual(invoice_item0[key], value)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_get("/billing/")
self.assert_not_in_success_response(
["Your plan will switch to annual billing on February 2, 2012"], response
@ -2792,10 +2794,10 @@ class StripeTest(StripeTestCase):
def test_reupgrade_after_plan_status_changed_to_downgrade_at_end_of_cycle(self) -> None:
user = self.example_user("hamlet")
self.login_user(user)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_patch(
"/json/billing/plan", {"status": CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE}
)
@ -2809,7 +2811,7 @@ class StripeTest(StripeTestCase):
assert plan is not None
self.assertEqual(plan.status, CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE)
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_patch("/json/billing/plan", {"status": CustomerPlan.ACTIVE})
expected_log = f"INFO:corporate.stripe:Change plan status: Customer.id: {stripe_customer_id}, CustomerPlan.id: {new_plan.id}, status: {CustomerPlan.ACTIVE}"
self.assertEqual(m.output[0], expected_log)
@ -2829,13 +2831,13 @@ class StripeTest(StripeTestCase):
# during the invoicing process.
user = self.example_user("hamlet")
self.login_user(user)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
with self.assertLogs("corporate.stripe", "INFO") as m:
stripe_customer_id = Customer.objects.get(realm=user.realm).id
new_plan = get_current_plan_by_realm(user.realm)
assert new_plan is not None
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.client_patch(
"/json/billing/plan", {"status": CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE}
)
@ -2859,7 +2861,7 @@ class StripeTest(StripeTestCase):
free_trial_end_date = self.now + timedelta(days=60)
with self.settings(FREE_TRIAL_DAYS=60):
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.add_card_and_upgrade(user, schedule="monthly")
plan = CustomerPlan.objects.get()
self.assertEqual(plan.next_invoice_date, free_trial_end_date)
@ -2914,7 +2916,7 @@ class StripeTest(StripeTestCase):
free_trial_end_date = self.now + timedelta(days=60)
with self.settings(FREE_TRIAL_DAYS=60):
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.add_card_and_upgrade(user, schedule="annual")
plan = CustomerPlan.objects.get()
self.assertEqual(plan.next_invoice_date, free_trial_end_date)
@ -2966,7 +2968,7 @@ class StripeTest(StripeTestCase):
free_trial_end_date = self.now + timedelta(days=60)
with self.settings(FREE_TRIAL_DAYS=60):
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, False, True)
plan = CustomerPlan.objects.get()
@ -2985,7 +2987,7 @@ class StripeTest(StripeTestCase):
self.login_user(user)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.client_patch("/json/billing/plan", {"status": CustomerPlan.ENDED})
plan.refresh_from_db()
@ -3015,7 +3017,7 @@ class StripeTest(StripeTestCase):
free_trial_end_date = self.now + timedelta(days=60)
with self.settings(FREE_TRIAL_DAYS=60):
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, False, True)
plan = get_current_plan_by_realm(user.realm)
assert plan is not None
@ -3027,7 +3029,7 @@ class StripeTest(StripeTestCase):
# Schedule downgrade
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_patch(
"/json/billing/plan",
{"status": CustomerPlan.DOWNGRADE_AT_END_OF_FREE_TRIAL},
@ -3045,7 +3047,7 @@ class StripeTest(StripeTestCase):
self.assertEqual(plan.licenses(), self.seat_count)
self.assertEqual(plan.licenses_at_next_renewal(), None)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
mock_customer = Mock(email=user.delivery_email)
mock_customer.invoice_settings.default_payment_method = Mock(
spec=stripe.PaymentMethod, type=Mock()
@ -3123,7 +3125,7 @@ class StripeTest(StripeTestCase):
free_trial_end_date = self.now + timedelta(days=60)
with self.settings(FREE_TRIAL_DAYS=60):
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, False, True)
plan = get_current_plan_by_realm(user.realm)
assert plan is not None
@ -3135,7 +3137,7 @@ class StripeTest(StripeTestCase):
# Schedule downgrade
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_patch(
"/json/billing/plan",
{"status": CustomerPlan.DOWNGRADE_AT_END_OF_FREE_TRIAL},
@ -3155,7 +3157,7 @@ class StripeTest(StripeTestCase):
# Cancel downgrade
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_patch(
"/json/billing/plan", {"status": CustomerPlan.FREE_TRIAL}
)
@ -3175,12 +3177,12 @@ class StripeTest(StripeTestCase):
def test_reupgrade_by_billing_admin_after_downgrade(self) -> None:
user = self.example_user("hamlet")
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
self.login_user(user)
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.client_patch(
"/json/billing/plan", {"status": CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE}
)
@ -3193,7 +3195,7 @@ class StripeTest(StripeTestCase):
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 time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
self.assertEqual(
m.output[0],
@ -3209,7 +3211,7 @@ class StripeTest(StripeTestCase):
self.assertEqual(response.status_code, 302)
self.assertEqual("/plans/", response["Location"])
with patch("corporate.lib.stripe.timezone_now", return_value=self.next_year):
with time_machine.travel(self.next_year, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
self.assertEqual(Customer.objects.count(), 1)
@ -3232,41 +3234,41 @@ class StripeTest(StripeTestCase):
user = self.example_user("hamlet")
self.login_user(user)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.upgrade(invoice=True, licenses=100)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
result = self.client_patch("/json/billing/plan", {"licenses": 100})
self.assert_json_error_contains(
result, "Your plan is already on 100 licenses in the current billing period."
)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
result = self.client_patch("/json/billing/plan", {"licenses_at_next_renewal": 100})
self.assert_json_error_contains(
result, "Your plan is already scheduled to renew with 100 licenses."
)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
result = self.client_patch("/json/billing/plan", {"licenses": 50})
self.assert_json_error_contains(
result, "You cannot decrease the licenses in the current billing period."
)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
result = self.client_patch("/json/billing/plan", {"licenses_at_next_renewal": 25})
self.assert_json_error_contains(
result,
"You must purchase licenses for all active users in your organization (minimum 30).",
)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
result = self.client_patch("/json/billing/plan", {"licenses": 2000})
self.assert_json_error_contains(
result, "Invoices with more than 1000 licenses can't be processed from this page."
)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
result = self.client_patch("/json/billing/plan", {"licenses": 150})
self.assert_json_success(result)
invoice_plans_as_needed(self.next_year)
@ -3316,7 +3318,7 @@ class StripeTest(StripeTestCase):
for key, value in line_item_params.items():
self.assertEqual(extra_license_item.get(key), value)
with patch("corporate.lib.stripe.timezone_now", return_value=self.next_year):
with time_machine.travel(self.next_year, tick=False):
result = self.client_patch("/json/billing/plan", {"licenses_at_next_renewal": 120})
self.assert_json_success(result)
invoice_plans_as_needed(self.next_year + timedelta(days=365))
@ -3366,10 +3368,10 @@ class StripeTest(StripeTestCase):
customer.exempt_from_license_number_check = True
customer.save()
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(100, False, CustomerPlan.ANNUAL, True, False)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
result = self.client_patch(
"/json/billing/plan",
{"licenses_at_next_renewal": get_latest_seat_count(user.realm) - 2},
@ -3400,7 +3402,7 @@ class StripeTest(StripeTestCase):
reduced_seat_count = get_latest_seat_count(user.realm) - 2
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(reduced_seat_count, False, CustomerPlan.ANNUAL, True, False)
latest_license_ledger = LicenseLedger.objects.last()
@ -3412,19 +3414,19 @@ class StripeTest(StripeTestCase):
user = self.example_user("hamlet")
self.login_user(user)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
result = self.client_patch("/json/billing/plan", {"licenses": 100})
self.assert_json_error_contains(result, "Your plan is on automatic license management.")
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
result = self.client_patch("/json/billing/plan", {"licenses_at_next_renewal": 100})
self.assert_json_error_contains(result, "Your plan is on automatic license management.")
def test_update_plan_with_invalid_status(self) -> None:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
self.login_user(self.example_user("hamlet"))
@ -3435,21 +3437,21 @@ class StripeTest(StripeTestCase):
self.assert_json_error_contains(response, "Invalid status")
def test_update_plan_without_any_params(self) -> None:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
self.login_user(self.example_user("hamlet"))
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
response = self.client_patch("/json/billing/plan", {})
self.assert_json_error_contains(response, "Nothing to change")
def test_update_plan_that_which_is_due_for_expiry(self) -> None:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
self.login_user(self.example_user("hamlet"))
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
result = self.client_patch(
"/json/billing/plan", {"status": CustomerPlan.DOWNGRADE_AT_END_OF_CYCLE}
)
@ -3459,19 +3461,19 @@ class StripeTest(StripeTestCase):
r"INFO:corporate.stripe:Change plan status: Customer.id: \d*, CustomerPlan.id: \d*, status: 2",
)
with patch("corporate.lib.stripe.timezone_now", return_value=self.next_year):
with time_machine.travel(self.next_year, tick=False):
result = self.client_patch("/json/billing/plan", {"status": CustomerPlan.ACTIVE})
self.assert_json_error_contains(
result, "Unable to update the plan. The plan has ended."
)
def test_update_plan_that_which_is_due_for_replacement(self) -> None:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.MONTHLY, True, False)
self.login_user(self.example_user("hamlet"))
with self.assertLogs("corporate.stripe", "INFO") as m:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
result = self.client_patch(
"/json/billing/plan", {"status": CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE}
)
@ -3481,7 +3483,7 @@ class StripeTest(StripeTestCase):
r"INFO:corporate.stripe:Change plan status: Customer.id: \d*, CustomerPlan.id: \d*, status: 4",
)
with patch("corporate.lib.stripe.timezone_now", return_value=self.next_month):
with time_machine.travel(self.next_month, tick=False):
result = self.client_patch("/json/billing/plan", {})
self.assert_json_error_contains(
result,
@ -3491,7 +3493,7 @@ class StripeTest(StripeTestCase):
@patch("corporate.lib.stripe.billing_logger.info")
def test_deactivate_realm(self, mock_: Mock) -> None:
user = self.example_user("hamlet")
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
plan = CustomerPlan.objects.get()
@ -3537,7 +3539,7 @@ class StripeTest(StripeTestCase):
def test_reupgrade_by_billing_admin_after_realm_deactivation(self) -> None:
user = self.example_user("hamlet")
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
do_deactivate_realm(get_realm("zulip"), acting_user=None)
@ -3549,7 +3551,7 @@ class StripeTest(StripeTestCase):
self.assertEqual(response.status_code, 302)
self.assertEqual("/plans/", response["Location"])
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
self.assertEqual(Customer.objects.count(), 1)
@ -4261,7 +4263,7 @@ class BillingHelpersTest(ZulipTestCase):
(anchor, month_later, month_later, 102),
),
]
with patch("corporate.lib.stripe.timezone_now", return_value=anchor):
with time_machine.travel(anchor, tick=False):
for (tier, automanage_licenses, billing_schedule, discount), output in test_cases:
output_ = compute_plan_parameters(
tier,
@ -4519,7 +4521,7 @@ class AnalyticsHelpersTest(ZulipTestCase):
class LicenseLedgerTest(StripeTestCase):
def test_add_plan_renewal_if_needed(self) -> None:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
self.assertEqual(LicenseLedger.objects.count(), 1)
plan = CustomerPlan.objects.get()
@ -4579,7 +4581,7 @@ class LicenseLedgerTest(StripeTestCase):
def test_update_license_ledger_for_automanaged_plan(self) -> None:
realm = get_realm("zulip")
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
plan = CustomerPlan.objects.first()
assert plan is not None
@ -4628,7 +4630,7 @@ class LicenseLedgerTest(StripeTestCase):
def test_update_license_ledger_for_manual_plan(self) -> None:
realm = get_realm("zulip")
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count + 1, False, CustomerPlan.ANNUAL, True, False)
plan = get_current_plan_by_realm(realm)
@ -4736,7 +4738,7 @@ class InvoiceTest(StripeTestCase):
def test_invoice_plan(self, *mocks: Mock) -> None:
user = self.example_user("hamlet")
self.login_user(user)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.add_card_and_upgrade(user)
# Increase
with patch("corporate.lib.stripe.get_latest_seat_count", return_value=self.seat_count + 3):
@ -4803,7 +4805,7 @@ class InvoiceTest(StripeTestCase):
# Also tests charge_automatically=False
user = self.example_user("hamlet")
self.login_user(user)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.upgrade(invoice=True)
plan = CustomerPlan.objects.first()
assert plan is not None
@ -4830,7 +4832,7 @@ class InvoiceTest(StripeTestCase):
self.assertEqual(item.get(key), value)
def test_no_invoice_needed(self) -> None:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
plan = CustomerPlan.objects.first()
assert plan is not None
@ -4843,7 +4845,7 @@ class InvoiceTest(StripeTestCase):
self.assertEqual(plan.next_invoice_date, self.next_month + timedelta(days=29))
def test_invoice_plans_as_needed(self) -> None:
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.local_upgrade(self.seat_count, True, CustomerPlan.ANNUAL, True, False)
plan = CustomerPlan.objects.first()
assert plan is not None
@ -5016,7 +5018,7 @@ class TestSupportBillingHelpers(StripeTestCase):
plan.status = CustomerPlan.ENDED
plan.save(update_fields=["status"])
attach_discount_to_realm(user.realm, Decimal(25), acting_user=support_admin)
with patch("corporate.lib.stripe.timezone_now", return_value=self.now):
with time_machine.travel(self.now, tick=False):
self.add_card_and_upgrade(
user, license_management="automatic", billing_modality="charge_automatically"
)

13
tools/semgrep-py.yml
View File

@ -282,3 +282,16 @@ rules:
Specify timedelta with named arguments.
languages: [python]
severity: ERROR
- id: time-machine
languages: [python]
patterns:
- pattern-either:
- pattern: patch("$FUNCTION", return_value=$TIME)
- pattern: mock.patch("$FUNCTION", return_value=$TIME)
- metavariable-regex:
metavariable: $FUNCTION
regex: .*timezone_now
fix: time_machine.travel($TIME, tick=False)
severity: ERROR
message: "Use the time_machine package, rather than mocking timezone_now"

3
zerver/tests/test_auth_backends.py
View File

@ -32,6 +32,7 @@ import ldap
import orjson
import requests
import responses
import time_machine
from bs4 import BeautifulSoup
from bs4.element import Tag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
@ -1931,7 +1932,7 @@ class SocialAuthBase(DesktopFlowTestingLib, ZulipTestCase, ABC):
result = self.social_auth_test(
account_data_dict, expect_choose_email_screen=True, subdomain=subdomain, is_signup=True
)
with mock.patch("zerver.models.timezone_now", return_value=now):
with time_machine.travel(now, tick=False):
self.stage_two_of_registration(
result, realm, subdomain, email, name, name, self.BACKEND_CLASS.full_name_validated
)

5
zerver/tests/test_digest.py
View File

@ -3,6 +3,7 @@ import time
from typing import List, Set
from unittest import mock
import time_machine
from django.test import override_settings
from django.utils.timezone import now as timezone_now
@ -451,7 +452,7 @@ class TestDigestEmailMessages(ZulipTestCase):
tuesday = self.tuesday()
cutoff = tuesday - datetime.timedelta(days=5)
with mock.patch("zerver.lib.digest.timezone_now", return_value=tuesday):
with time_machine.travel(tuesday, tick=False):
with mock.patch("zerver.lib.digest.queue_digest_user_ids") as queue_mock:
enqueue_emails(cutoff)
queue_mock.assert_not_called()
@ -463,7 +464,7 @@ class TestDigestEmailMessages(ZulipTestCase):
not_tuesday = datetime.datetime(year=2016, month=1, day=6, tzinfo=datetime.timezone.utc)
cutoff = not_tuesday - datetime.timedelta(days=5)
with mock.patch("zerver.lib.digest.timezone_now", return_value=not_tuesday):
with time_machine.travel(not_tuesday, tick=False):
with mock.patch("zerver.lib.digest.queue_digest_user_ids") as queue_mock:
enqueue_emails(cutoff)
queue_mock.assert_not_called()

4
zerver/tests/test_email_change.py
View File

@ -1,7 +1,7 @@
import datetime
from email.headerregistry import Address
from unittest import mock
import time_machine
from django.conf import settings
from django.core import mail
from django.utils.html import escape
@ -79,7 +79,7 @@ class EmailChangeTestCase(ZulipTestCase):
realm=user_profile.realm,
)
date_sent = now() - datetime.timedelta(days=2)
with mock.patch("confirmation.models.timezone_now", return_value=date_sent):
with time_machine.travel(date_sent, tick=False):
url = create_confirmation_link(obj, Confirmation.EMAIL_CHANGE)
response = self.client_get(url)

6
zerver/tests/test_example.py
View File

@ -2,6 +2,7 @@ import datetime
from unittest import mock
import orjson
import time_machine
from django.utils.timezone import now as timezone_now
from zerver.actions.users import do_change_can_create_users, do_change_user_role
@ -491,10 +492,7 @@ class TestMocking(ZulipTestCase):
seconds=MESSAGE_CONTENT_EDIT_LIMIT + 100
) # There's a buffer time applied to the limit, hence the extra 100s.
with mock.patch(
"zerver.actions.message_edit.timezone_now",
return_value=time_beyond_edit_limit,
):
with time_machine.travel(time_beyond_edit_limit, tick=False):
result = self.client_patch(
f"/json/messages/{sent_message_id}", {"content": "I actually want pizza."}
)

6
zerver/tests/test_gitter_importer.py
View File

@ -5,6 +5,7 @@ from unittest import mock
import dateutil.parser
import orjson
import time_machine
from zerver.data_import.gitter import do_convert_data, get_usermentions
from zerver.lib.import_realm import do_import_realm
@ -33,9 +34,8 @@ class GitterImporter(ZulipTestCase):
with open(gitter_file) as f:
gitter_data = orjson.loads(f.read())
sent_datetime = dateutil.parser.parse(gitter_data[1]["sent"])
with self.assertLogs(level="INFO"), mock.patch(
"zerver.data_import.import_util.timezone_now",
return_value=sent_datetime + timedelta(days=1),
with self.assertLogs(level="INFO"), time_machine.travel(
(sent_datetime + timedelta(days=1)), tick=False
):
do_convert_data(gitter_file, output_dir)

7
zerver/tests/test_home.py
View File

@ -6,6 +6,7 @@ from typing import TYPE_CHECKING, Any, Dict
from unittest.mock import patch
import orjson
import time_machine
from django.conf import settings
from django.test import override_settings
from django.utils.timezone import now as timezone_now
@ -1020,17 +1021,17 @@ class HomeTest(ZulipTestCase):
hamlet = self.example_user("hamlet")
iago = self.example_user("iago")
now = LAST_SERVER_UPGRADE_TIME.replace(tzinfo=datetime.timezone.utc)
with patch("zerver.lib.compatibility.timezone_now", return_value=now + timedelta(days=10)):
with time_machine.travel((now + timedelta(days=10)), tick=False):
self.assertEqual(is_outdated_server(iago), False)
self.assertEqual(is_outdated_server(hamlet), False)
self.assertEqual(is_outdated_server(None), False)
with patch("zerver.lib.compatibility.timezone_now", return_value=now + timedelta(days=397)):
with time_machine.travel((now + timedelta(days=397)), tick=False):
self.assertEqual(is_outdated_server(iago), True)
self.assertEqual(is_outdated_server(hamlet), True)
self.assertEqual(is_outdated_server(None), True)
with patch("zerver.lib.compatibility.timezone_now", return_value=now + timedelta(days=380)):
with time_machine.travel((now + timedelta(days=380)), tick=False):
self.assertEqual(is_outdated_server(iago), True)
self.assertEqual(is_outdated_server(hamlet), False)
self.assertEqual(is_outdated_server(None), False)

14
zerver/tests/test_invite.py
View File

@ -1439,7 +1439,7 @@ so we didn't send them an invitation. We did send invitations to everyone else!"
email=email, referred_by=inviter, realm=realm
)
date_sent = timezone_now() - datetime.timedelta(weeks=3)
with patch("confirmation.models.timezone_now", return_value=date_sent):
with time_machine.travel(date_sent, tick=False):
url = create_confirmation_link(prereg_user, Confirmation.USER_REGISTRATION)
key = url.split("/")[-1]
@ -1782,10 +1782,7 @@ class InvitationsTestCase(InviteUserBase):
invite_expires_in_minutes=invite_expires_in_minutes,
)
with patch(
"confirmation.models.timezone_now",
return_value=timezone_now() - datetime.timedelta(days=3),
):
with time_machine.travel((timezone_now() - datetime.timedelta(days=3)), tick=False):
do_invite_users(
user_profile,
["TestTwo@zulip.com"],
@ -1828,10 +1825,7 @@ class InvitationsTestCase(InviteUserBase):
get_stream(stream_name, user_profile.realm) for stream_name in ["Denmark", "Scotland"]
]
with patch(
"confirmation.models.timezone_now",
return_value=timezone_now() - datetime.timedelta(days=1000),
):
with time_machine.travel((timezone_now() - datetime.timedelta(days=1000)), tick=False):
# Testing the invitation with expiry date set to "None" exists
# after a large amount of days.
do_invite_users(
@ -2291,7 +2285,7 @@ class MultiuseInviteTest(ZulipTestCase):
if date_sent is None:
date_sent = timezone_now()
validity_in_minutes = 2 * 24 * 60
with patch("confirmation.models.timezone_now", return_value=date_sent):
with time_machine.travel(date_sent, tick=False):
return create_confirmation_link(
invite, Confirmation.MULTIUSE_INVITE, validity_in_minutes=validity_in_minutes
)

10
zerver/tests/test_muted_users.py
View File

@ -1,6 +1,8 @@
from datetime import datetime, timezone
from unittest import mock
import time_machine
from zerver.actions.users import do_deactivate_user
from zerver.lib.cache import cache_get, get_muting_users_cache_key
from zerver.lib.muted_users import get_mute_object, get_muting_users, get_user_mutes
@ -19,7 +21,7 @@ class MutedUsersTests(ZulipTestCase):
self.assertEqual(muted_users, [])
mute_time = datetime(2021, 1, 1, tzinfo=timezone.utc)
with mock.patch("zerver.views.muted_users.timezone_now", return_value=mute_time):
with time_machine.travel(mute_time, tick=False):
url = f"/api/v1/users/me/muted_users/{cordelia.id}"
result = self.api_post(hamlet, url)
self.assert_json_success(result)
@ -85,7 +87,7 @@ class MutedUsersTests(ZulipTestCase):
if deactivate_user:
do_deactivate_user(cordelia, acting_user=None)
with mock.patch("zerver.views.muted_users.timezone_now", return_value=mute_time):
with time_machine.travel(mute_time, tick=False):
url = f"/api/v1/users/me/muted_users/{cordelia.id}"
result = self.api_post(hamlet, url)
self.assert_json_success(result)
@ -139,12 +141,12 @@ class MutedUsersTests(ZulipTestCase):
if deactivate_user:
do_deactivate_user(cordelia, acting_user=None)
with mock.patch("zerver.views.muted_users.timezone_now", return_value=mute_time):
with time_machine.travel(mute_time, tick=False):
url = f"/api/v1/users/me/muted_users/{cordelia.id}"
result = self.api_post(hamlet, url)
self.assert_json_success(result)
with mock.patch("zerver.actions.muted_users.timezone_now", return_value=mute_time):
with time_machine.travel(mute_time, tick=False):
# To test that `RealmAuditLog` entry has correct `event_time`.
url = f"/api/v1/users/me/muted_users/{cordelia.id}"
result = self.api_delete(hamlet, url)

10
zerver/tests/test_new_users.py
View File

@ -1,8 +1,8 @@
import datetime
import sys
from typing import Sequence
from unittest import mock
import time_machine
from django.conf import settings
from django.core import mail
from django.test import override_settings
@ -55,7 +55,7 @@ class SendLoginEmailTest(ZulipTestCase):
user_tz = zoneinfo.ZoneInfo(user.timezone)
mock_time = datetime.datetime(year=2018, month=1, day=1, tzinfo=datetime.timezone.utc)
reference_time = mock_time.astimezone(user_tz).strftime("%A, %B %d, %Y at %I:%M %p %Z")
with mock.patch("zerver.signals.timezone_now", return_value=mock_time):
with time_machine.travel(mock_time, tick=False):
self.client_post(
"/accounts/login/", info=login_info, HTTP_USER_AGENT=firefox_windows
)
@ -71,7 +71,7 @@ class SendLoginEmailTest(ZulipTestCase):
self.logout() # We just logged in, we'd be redirected without this
user.twenty_four_hour_time = True
user.save()
with mock.patch("zerver.signals.timezone_now", return_value=mock_time):
with time_machine.travel(mock_time, tick=False):
self.client_post(
"/accounts/login/", info=login_info, HTTP_USER_AGENT=firefox_windows
)
@ -116,13 +116,13 @@ class SendLoginEmailTest(ZulipTestCase):
do_change_user_setting(user, "enable_login_emails", False, acting_user=None)
self.assertFalse(user.enable_login_emails)
with mock.patch("zerver.signals.timezone_now", return_value=mock_time):
with time_machine.travel(mock_time, tick=False):
self.login_user(user)
self.assert_length(mail.outbox, 0)
do_change_user_setting(user, "enable_login_emails", True, acting_user=None)
self.assertTrue(user.enable_login_emails)
with mock.patch("zerver.signals.timezone_now", return_value=mock_time):
with time_machine.travel(mock_time, tick=False):
self.login_user(user)
self.assert_length(mail.outbox, 1)

33
zerver/tests/test_presence.py
View File

@ -3,6 +3,7 @@ from datetime import timedelta
from typing import Any, Dict
from unittest import mock
import time_machine
from django.conf import settings
from django.utils.timezone import now as timezone_now
from typing_extensions import override
@ -238,7 +239,7 @@ class UserPresenceTests(ZulipTestCase):
self.login("hamlet")
self.assertEqual(UserActivityInterval.objects.filter(user_profile=user_profile).count(), 0)
time_zero = timezone_now().replace(microsecond=0)
with mock.patch("zerver.views.presence.timezone_now", return_value=time_zero):
with time_machine.travel(time_zero, tick=False):
result = self.client_post(
"/json/users/me/presence", {"status": "active", "new_user_input": "true"}
)
@ -250,7 +251,7 @@ class UserPresenceTests(ZulipTestCase):
second_time = time_zero + timedelta(seconds=600)
# Extent the interval
with mock.patch("zerver.views.presence.timezone_now", return_value=second_time):
with time_machine.travel(second_time, tick=False):
result = self.client_post(
"/json/users/me/presence", {"status": "active", "new_user_input": "true"}
)
@ -261,7 +262,7 @@ class UserPresenceTests(ZulipTestCase):
self.assertEqual(interval.end, second_time + UserActivityInterval.MIN_INTERVAL_LENGTH)
third_time = time_zero + timedelta(seconds=6000)
with mock.patch("zerver.views.presence.timezone_now", return_value=third_time):
with time_machine.travel(third_time, tick=False):
result = self.client_post(
"/json/users/me/presence", {"status": "active", "new_user_input": "true"}
)
@ -529,22 +530,21 @@ class UserPresenceAggregationTests(ZulipTestCase):
self, user: UserProfile, status: str, validate_time: datetime.datetime
) -> Dict[str, Dict[str, Any]]:
self.login_user(user)
timezone_util = "zerver.views.presence.timezone_now"
# First create some initial, old presence to avoid the details of the edge case of initial
# presence creation messing with the intended setup.
with mock.patch(timezone_util, return_value=validate_time - datetime.timedelta(days=365)):
with time_machine.travel((validate_time - datetime.timedelta(days=365)), tick=False):
self.client_post("/json/users/me/presence", {"status": status})
with mock.patch(timezone_util, return_value=validate_time - datetime.timedelta(seconds=5)):
with time_machine.travel((validate_time - datetime.timedelta(seconds=5)), tick=False):
self.client_post("/json/users/me/presence", {"status": status})
with mock.patch(timezone_util, return_value=validate_time - datetime.timedelta(seconds=2)):
with time_machine.travel((validate_time - datetime.timedelta(seconds=2)), tick=False):
self.api_post(
user,
"/api/v1/users/me/presence",
{"status": status},
HTTP_USER_AGENT="ZulipAndroid/1.0",
)
with mock.patch(timezone_util, return_value=validate_time - datetime.timedelta(seconds=7)):
with time_machine.travel((validate_time - datetime.timedelta(seconds=7)), tick=False):
latest_result = self.api_post(
user,
"/api/v1/users/me/presence",
@ -571,10 +571,7 @@ class UserPresenceAggregationTests(ZulipTestCase):
offset = datetime.timedelta(seconds=settings.PRESENCE_UPDATE_MIN_FREQ_SECONDS + 1)
validate_time = timezone_now() - offset
self._send_presence_for_aggregated_tests(user, "active", validate_time)
with mock.patch(
"zerver.views.presence.timezone_now",
return_value=validate_time + offset,
):
with time_machine.travel((validate_time + offset), tick=False):
result = self.api_post(
user,
"/api/v1/users/me/presence",
@ -620,10 +617,7 @@ class UserPresenceAggregationTests(ZulipTestCase):
self.login_user(user)
validate_time = timezone_now()
self._send_presence_for_aggregated_tests(user, "idle", validate_time)
with mock.patch(
"zerver.views.presence.timezone_now",
return_value=validate_time - datetime.timedelta(seconds=3),
):
with time_machine.travel((validate_time - datetime.timedelta(seconds=3)), tick=False):
result_dict = self.api_post(
user,
"/api/v1/users/me/presence",
@ -646,10 +640,9 @@ class UserPresenceAggregationTests(ZulipTestCase):
validate_time = timezone_now()
result_dict = self._send_presence_for_aggregated_tests(user, "idle", validate_time)
with mock.patch(
"zerver.views.presence.timezone_now",
return_value=validate_time
+ datetime.timedelta(seconds=settings.OFFLINE_THRESHOLD_SECS + 1),
with time_machine.travel(
(validate_time + datetime.timedelta(seconds=settings.OFFLINE_THRESHOLD_SECS + 1)),
tick=False,
):
# After settings.OFFLINE_THRESHOLD_SECS + 1 this generated, recent presence data
# will count as offline.

8
zerver/tests/test_sessions.py
View File

@ -1,7 +1,7 @@
from datetime import timedelta
from typing import Any, Callable
from unittest import mock
import time_machine
from django.utils.timezone import now as timezone_now
from typing_extensions import override
@ -134,15 +134,13 @@ class TestExpirableSessionVars(ZulipTestCase):
def test_set_and_get_basic(self) -> None:
start_time = timezone_now()
with mock.patch("zerver.lib.sessions.timezone_now", return_value=start_time):
with time_machine.travel(start_time, tick=False):
set_expirable_session_var(
self.session, "test_set_and_get_basic", "some_value", expiry_seconds=10
)
value = get_expirable_session_var(self.session, "test_set_and_get_basic")
self.assertEqual(value, "some_value")
with mock.patch(
"zerver.lib.sessions.timezone_now", return_value=start_time + timedelta(seconds=11)
):
with time_machine.travel((start_time + timedelta(seconds=11)), tick=False):
value = get_expirable_session_var(self.session, "test_set_and_get_basic")
self.assertEqual(value, None)

5
zerver/tests/test_user_groups.py
View File

@ -3,6 +3,7 @@ from typing import Iterable, Optional
from unittest import mock
import orjson
import time_machine
from django.db import transaction
from django.utils.timezone import now as timezone_now
@ -1189,9 +1190,7 @@ class UserGroupAPITestCase(UserGroupTestCase):
)
current_time = timezone_now()
with mock.patch(
"zerver.actions.user_groups.timezone_now", return_value=current_time + timedelta(days=3)
):
with time_machine.travel((current_time + timedelta(days=3)), tick=False):
promote_new_full_members()
self.assertTrue(