mirror of
				https://github.com/zulip/zulip.git
				synced 2025-10-31 12:03:46 +00:00 
			
		
		
		
	This commit updates code to not pre-fetch DM permission group settings using select_related and instead just fetch the required data from DB when checking permission. This will increase one query but will help in pre-fetching the settings for all users and for all type of messages. Fixes part of #33677.
		
			
				
	
	
		
			572 lines
		
	
	
		
			24 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			572 lines
		
	
	
		
			24 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
| from datetime import timedelta
 | |
| from unittest import mock
 | |
| 
 | |
| import orjson
 | |
| import time_machine
 | |
| from django.utils.timezone import now as timezone_now
 | |
| 
 | |
| from zerver.actions.realm_settings import do_change_realm_permission_group_setting
 | |
| from zerver.actions.users import do_change_can_create_users, do_change_user_role
 | |
| from zerver.lib.exceptions import JsonableError, StreamWildcardMentionNotAllowedError
 | |
| from zerver.lib.streams import access_stream_for_send_message
 | |
| from zerver.lib.test_classes import ZulipTestCase
 | |
| from zerver.lib.test_helpers import most_recent_message
 | |
| from zerver.lib.users import is_administrator_role
 | |
| from zerver.models import Realm, UserProfile, UserStatus
 | |
| from zerver.models.groups import NamedUserGroup, SystemGroups
 | |
| from zerver.models.realms import get_realm
 | |
| from zerver.models.streams import get_stream
 | |
| from zerver.models.users import get_user_by_delivery_email
 | |
| 
 | |
| 
 | |
| # Most Zulip tests use ZulipTestCase, which inherits from django.test.TestCase.
 | |
| # We recommend learning Django basics first, so search the web for "django testing".
 | |
| # A common first result is https://docs.djangoproject.com/en/5.0/topics/testing/
 | |
| class TestBasics(ZulipTestCase):
 | |
|     def test_basics(self) -> None:
 | |
|         # Django's tests are based on Python's unittest module, so you
 | |
|         # will see us use things like assertEqual, assertTrue, and assertRaisesRegex
 | |
|         # quite often.
 | |
|         # See https://docs.python.org/3/library/unittest.html#unittest.TestCase.assertEqual
 | |
|         self.assertEqual(7 * 6, 42)
 | |
| 
 | |
| 
 | |
| class TestBasicUserStuff(ZulipTestCase):
 | |
|     # Zulip has test fixtures with built-in users.  It's good to know
 | |
|     # which users are special. For example, Iago is our built-in
 | |
|     # realm administrator.  You can also modify users as needed.
 | |
|     def test_users(self) -> None:
 | |
|         # The example_user() helper returns a UserProfile object.
 | |
|         hamlet = self.example_user("hamlet")
 | |
|         self.assertEqual(hamlet.full_name, "King Hamlet")
 | |
|         self.assertEqual(hamlet.role, UserProfile.ROLE_MEMBER)
 | |
| 
 | |
|         iago = self.example_user("iago")
 | |
|         self.assertEqual(iago.role, UserProfile.ROLE_REALM_ADMINISTRATOR)
 | |
| 
 | |
|         polonius = self.example_user("polonius")
 | |
|         self.assertEqual(polonius.role, UserProfile.ROLE_GUEST)
 | |
| 
 | |
|         self.assertEqual(self.example_email("cordelia"), "cordelia@zulip.com")
 | |
| 
 | |
|     def test_lib_functions(self) -> None:
 | |
|         # This test is an example of testing a single library function.
 | |
|         # Our tests aren't always at this level of granularity, but it's
 | |
|         # often possible to write concise tests for library functions.
 | |
| 
 | |
|         # Get our UserProfile objects first.
 | |
|         iago = self.example_user("iago")
 | |
|         hamlet = self.example_user("hamlet")
 | |
| 
 | |
|         # It is a good idea for your tests to clearly demonstrate a
 | |
|         # **change** to a value.  So here we want to make sure that
 | |
|         # do_change_user_role will change Hamlet such that
 | |
|         # is_administrator_role becomes True, but we first assert it's
 | |
|         # False.
 | |
|         self.assertFalse(is_administrator_role(hamlet.role))
 | |
| 
 | |
|         # Tests should modify properties using the standard library
 | |
|         # functions, like do_change_user_role. Modifying Django
 | |
|         # objects and then using .save() can be buggy, as doing so can
 | |
|         # fail to update caches, RealmAuditLog, or related tables properly.
 | |
|         do_change_user_role(hamlet, UserProfile.ROLE_REALM_OWNER, acting_user=iago)
 | |
|         self.assertTrue(is_administrator_role(hamlet.role))
 | |
| 
 | |
|         # After we promote Hamlet, we also demote him.  Testing state
 | |
|         # changes like this in a single test can be a good technique,
 | |
|         # although we also don't want tests to be too long.
 | |
|         #
 | |
|         # Important note: You don't need to undo changes done in the
 | |
|         # test at the end. Every test is run inside a database
 | |
|         # transaction, that is reverted after the test completes.
 | |
|         # There are a few exceptions, where tests interact with the
 | |
|         # filesystem (E.g. uploading files), which is generally
 | |
|         # handled by the setUp/tearDown methods for the test class.
 | |
|         do_change_user_role(hamlet, UserProfile.ROLE_MODERATOR, acting_user=iago)
 | |
|         self.assertFalse(is_administrator_role(hamlet.role))
 | |
| 
 | |
| 
 | |
| class TestFullStack(ZulipTestCase):
 | |
|     # Zulip's backend tests are largely full-stack integration tests,
 | |
|     # making use of some strategic mocking at times, though we do use
 | |
|     # unit tests for some classes of low-level functions.
 | |
|     #
 | |
|     # See https://zulip.readthedocs.io/en/latest/testing/philosophy.html
 | |
|     # for details on this and other testing design decisions.
 | |
|     def test_client_get(self) -> None:
 | |
|         hamlet = self.example_user("hamlet")
 | |
|         cordelia = self.example_user("cordelia")
 | |
| 
 | |
|         # Most full-stack tests require you to log in the user.
 | |
|         # The login_user helper basically wraps Django's client.login().
 | |
|         self.login_user(hamlet)
 | |
| 
 | |
|         # Zulip's client_get is a very thin wrapper on Django's client.get.
 | |
|         # We always use the Zulip wrappers for client_get and client_post.
 | |
|         url = f"/json/users/{cordelia.id}"
 | |
|         result = self.client_get(url)
 | |
| 
 | |
|         # Almost every meaningful full-stack test for a "happy path" situation
 | |
|         # uses assert_json_success().
 | |
|         self.assert_json_success(result)
 | |
| 
 | |
|         # When we unpack the result.content object, we prefer the orjson library.
 | |
|         content = orjson.loads(result.content)
 | |
| 
 | |
|         # In this case we will validate the entire payload. It's good to use
 | |
|         # concrete values where possible, but some things, like "cordelia.id",
 | |
|         # are somewhat unpredictable, so we don't hard code values.
 | |
|         #
 | |
|         # Others, like email and full_name here, are fields we haven't
 | |
|         # changed, and thus explicit values would just be hardcoding
 | |
|         # test database defaults in additional places.
 | |
|         self.assertEqual(
 | |
|             content["user"],
 | |
|             dict(
 | |
|                 avatar_url=content["user"]["avatar_url"],
 | |
|                 avatar_version=1,
 | |
|                 date_joined=content["user"]["date_joined"],
 | |
|                 delivery_email=None,
 | |
|                 email=cordelia.email,
 | |
|                 full_name=cordelia.full_name,
 | |
|                 is_active=True,
 | |
|                 is_admin=False,
 | |
|                 is_bot=False,
 | |
|                 is_guest=False,
 | |
|                 is_owner=False,
 | |
|                 role=UserProfile.ROLE_MEMBER,
 | |
|                 timezone="Etc/UTC",
 | |
|                 user_id=cordelia.id,
 | |
|             ),
 | |
|         )
 | |
| 
 | |
|     def test_client_post(self) -> None:
 | |
|         # Here we're gonna test a POST call to /json/users, and it's
 | |
|         # important that we not only check the payload, but we make
 | |
|         # sure that the intended side effects actually happen.
 | |
|         iago = self.example_user("iago")
 | |
|         self.login_user(iago)
 | |
| 
 | |
|         realm = get_realm("zulip")
 | |
|         self.assertEqual(realm.id, iago.realm_id)
 | |
| 
 | |
|         # Get our failing test first.
 | |
|         self.assertRaises(
 | |
|             UserProfile.DoesNotExist, lambda: get_user_by_delivery_email("romeo@zulip.net", realm)
 | |
|         )
 | |
| 
 | |
|         # Before we can successfully post, we need to ensure
 | |
|         # that Iago can create users.
 | |
|         do_change_can_create_users(iago, True)
 | |
| 
 | |
|         params = dict(
 | |
|             email="romeo@zulip.net",
 | |
|             password="xxxx",
 | |
|             full_name="Romeo Montague",
 | |
|         )
 | |
| 
 | |
|         # Use the Zulip wrapper.
 | |
|         result = self.client_post("/json/users", params)
 | |
| 
 | |
|         # Once again we check that the HTTP request was successful.
 | |
|         self.assert_json_success(result)
 | |
|         content = orjson.loads(result.content)
 | |
| 
 | |
|         # Finally we test the side effect of the post.
 | |
|         user_id = content["user_id"]
 | |
|         romeo = get_user_by_delivery_email("romeo@zulip.net", realm)
 | |
|         self.assertEqual(romeo.id, user_id)
 | |
| 
 | |
|     def test_can_create_users(self) -> None:
 | |
|         # Typically, when testing an API endpoint, we prefer a single
 | |
|         # test covering both the happy path and common error paths.
 | |
|         #
 | |
|         # See https://zulip.readthedocs.io/en/latest/testing/philosophy.html#share-test-setup-code.
 | |
|         iago = self.example_user("iago")
 | |
|         self.login_user(iago)
 | |
| 
 | |
|         do_change_can_create_users(iago, False)
 | |
|         valid_params = dict(
 | |
|             email="romeo@zulip.net",
 | |
|             password="xxxx",
 | |
|             full_name="Romeo Montague",
 | |
|         )
 | |
| 
 | |
|         # We often use assert_json_error for negative tests.
 | |
|         result = self.client_post("/json/users", valid_params)
 | |
|         self.assert_json_error(result, "User not authorized to create users", 400)
 | |
| 
 | |
|         do_change_can_create_users(iago, True)
 | |
|         incomplete_params = dict(
 | |
|             full_name="Romeo Montague",
 | |
|         )
 | |
|         result = self.client_post("/json/users", incomplete_params)
 | |
|         self.assert_json_error(result, "Missing 'email' argument", 400)
 | |
| 
 | |
|         # Verify that the original parameters were valid. Especially
 | |
|         # for errors with generic error messages, this is important to
 | |
|         # confirm that the original request with these parameters
 | |
|         # failed because of incorrect permissions, and not because
 | |
|         # valid_params weren't actually valid.
 | |
|         result = self.client_post("/json/users", valid_params)
 | |
|         self.assert_json_success(result)
 | |
| 
 | |
|         # Verify error handling when the user already exists.
 | |
|         result = self.client_post("/json/users", valid_params)
 | |
|         self.assert_json_error(result, "Email is already in use.", 400)
 | |
| 
 | |
|     def test_tornado_redirects(self) -> None:
 | |
|         # Let's poke a bit at Zulip's event system.
 | |
|         # See https://zulip.readthedocs.io/en/latest/subsystems/events-system.html
 | |
|         # for context on the system itself and how it should be tested.
 | |
|         #
 | |
|         # Most specific features that might feel tricky to test have
 | |
|         # similarly handy helpers, so find similar tests with `git grep` and read them!
 | |
|         cordelia = self.example_user("cordelia")
 | |
|         self.login_user(cordelia)
 | |
| 
 | |
|         params = dict(status_text="on vacation")
 | |
| 
 | |
|         # Use the capture_send_event_calls context manager to capture events.
 | |
|         with self.capture_send_event_calls(expected_num_events=1) as events:
 | |
|             result = self.api_post(cordelia, "/api/v1/users/me/status", params)
 | |
| 
 | |
|         self.assert_json_success(result)
 | |
| 
 | |
|         # Check that the POST to Zulip caused the expected events to be sent
 | |
|         # to Tornado.
 | |
|         self.assertEqual(
 | |
|             events[0]["event"],
 | |
|             dict(type="user_status", user_id=cordelia.id, status_text="on vacation"),
 | |
|         )
 | |
| 
 | |
|         # Grabbing the last row in the table is OK here, but often it's
 | |
|         # better to look up the object we created via its ID,
 | |
|         # especially if there's risk of similar objects existing
 | |
|         # (E.g. a message sent to that topic earlier in the test).
 | |
|         row = UserStatus.objects.last()
 | |
|         assert row is not None
 | |
|         self.assertEqual(row.user_profile_id, cordelia.id)
 | |
|         self.assertEqual(row.status_text, "on vacation")
 | |
| 
 | |
| 
 | |
| class TestStreamHelpers(ZulipTestCase):
 | |
|     # Streams are an important concept in Zulip, and ZulipTestCase
 | |
|     # has helpers such as subscribe, users_subscribed_to_stream,
 | |
|     # and make_stream.
 | |
|     def test_new_streams(self) -> None:
 | |
|         cordelia = self.example_user("cordelia")
 | |
|         othello = self.example_user("othello")
 | |
|         realm = cordelia.realm
 | |
| 
 | |
|         stream_name = "Some new stream"
 | |
|         self.subscribe(cordelia, stream_name)
 | |
| 
 | |
|         self.assertEqual(set(self.users_subscribed_to_stream(stream_name, realm)), {cordelia})
 | |
| 
 | |
|         self.subscribe(othello, stream_name)
 | |
|         self.assertEqual(
 | |
|             set(self.users_subscribed_to_stream(stream_name, realm)), {cordelia, othello}
 | |
|         )
 | |
| 
 | |
|     def test_private_stream(self) -> None:
 | |
|         # When we test stream permissions, it's very common to use at least
 | |
|         # two users, so that you can see how different users are impacted.
 | |
|         # We commonly use Othello to represent the "other" user from the primary user.
 | |
|         cordelia = self.example_user("cordelia")
 | |
|         othello = self.example_user("othello")
 | |
| 
 | |
|         realm = cordelia.realm
 | |
|         stream_name = "Some private stream"
 | |
| 
 | |
|         # Use the invite_only flag in make_stream to make a stream "private".
 | |
|         stream = self.make_stream(stream_name=stream_name, invite_only=True)
 | |
|         self.subscribe(cordelia, stream_name)
 | |
| 
 | |
|         self.assertEqual(set(self.users_subscribed_to_stream(stream_name, realm)), {cordelia})
 | |
| 
 | |
|         stream = get_stream(stream_name, realm)
 | |
|         self.assertEqual(stream.name, stream_name)
 | |
|         self.assertTrue(stream.invite_only)
 | |
| 
 | |
|         # We will now observe that Cordelia can access the stream...
 | |
|         access_stream_for_send_message(cordelia, stream, forwarder_user_profile=None)
 | |
| 
 | |
|         # ...but Othello can't.
 | |
|         with self.assertRaisesRegex(JsonableError, "Not authorized to send to channel"):
 | |
|             access_stream_for_send_message(othello, stream, forwarder_user_profile=None)
 | |
| 
 | |
| 
 | |
| class TestMessageHelpers(ZulipTestCase):
 | |
|     # If you are testing behavior related to messages, then it's good
 | |
|     # to know about send_stream_message, send_personal_message, and
 | |
|     # most_recent_message.
 | |
|     def test_stream_message(self) -> None:
 | |
|         hamlet = self.example_user("hamlet")
 | |
|         iago = self.example_user("iago")
 | |
|         self.subscribe(hamlet, "Denmark")
 | |
|         self.subscribe(iago, "Denmark")
 | |
| 
 | |
|         # The functions to send a message return the ID of the created
 | |
|         # message, so you usually you don't need to look it up.
 | |
|         sent_message_id = self.send_stream_message(
 | |
|             sender=hamlet,
 | |
|             stream_name="Denmark",
 | |
|             topic_name="lunch",
 | |
|             content="I want pizza!",
 | |
|         )
 | |
| 
 | |
|         # But if you want to verify the most recent message received
 | |
|         # by a user, there's a handy function for that.
 | |
|         iago_message = most_recent_message(iago)
 | |
| 
 | |
|         # Here we check that the message we sent is the last one that
 | |
|         # Iago received.  While we verify several properties of the
 | |
|         # last message, the most important to verify is the unique ID,
 | |
|         # since that protects us from bugs if this test were to be
 | |
|         # extended to send multiple similar messages.
 | |
|         self.assertEqual(iago_message.id, sent_message_id)
 | |
|         self.assertEqual(iago_message.sender_id, hamlet.id)
 | |
|         self.assert_message_stream_name(iago_message, "Denmark")
 | |
|         self.assertEqual(iago_message.topic_name(), "lunch")
 | |
|         self.assertEqual(iago_message.content, "I want pizza!")
 | |
| 
 | |
|     def test_personal_message(self) -> None:
 | |
|         hamlet = self.example_user("hamlet")
 | |
|         cordelia = self.example_user("cordelia")
 | |
| 
 | |
|         sent_message_id = self.send_personal_message(
 | |
|             from_user=hamlet,
 | |
|             to_user=cordelia,
 | |
|             content="hello there!",
 | |
|         )
 | |
| 
 | |
|         cordelia_message = most_recent_message(cordelia)
 | |
| 
 | |
|         self.assertEqual(cordelia_message.id, sent_message_id)
 | |
|         self.assertEqual(cordelia_message.sender_id, hamlet.id)
 | |
|         self.assertEqual(cordelia_message.content, "hello there!")
 | |
| 
 | |
| 
 | |
| class TestQueryCounts(ZulipTestCase):
 | |
|     def test_capturing_queries(self) -> None:
 | |
|         # It's a common pitfall in Django to accidentally perform
 | |
|         # database queries in a loop, due to lazy evaluation of
 | |
|         # foreign keys. We use the assert_database_query_count
 | |
|         # context manager to ensure our query count is predictable.
 | |
|         #
 | |
|         # When a test containing one of these query count assertions
 | |
|         # fails, we'll want to understand the new queries and whether
 | |
|         # they're necessary. You can investiate whether the changes
 | |
|         # are expected/sensible by comparing print(queries) between
 | |
|         # your branch and main.
 | |
|         hamlet = self.example_user("hamlet")
 | |
|         cordelia = self.example_user("cordelia")
 | |
| 
 | |
|         with self.assert_database_query_count(16):
 | |
|             self.send_personal_message(
 | |
|                 from_user=hamlet,
 | |
|                 to_user=cordelia,
 | |
|                 content="hello there!",
 | |
|             )
 | |
| 
 | |
| 
 | |
| class TestDevelopmentEmailsLog(ZulipTestCase):
 | |
|     # The /emails/generate/ endpoint can be used to generate
 | |
|     # all sorts of emails. Those can be accessed at /emails/
 | |
|     # in development server. Let's test that here.
 | |
|     def test_generate_emails(self) -> None:
 | |
|         # It is a common case where some functions that we test rely
 | |
|         # on a certain setting's value. You can test those under the
 | |
|         # context of a desired setting value as done below.
 | |
|         #
 | |
|         # The endpoint we're testing here rely on these settings:
 | |
|         #   * EMAIL_BACKEND: The backend class used to send emails.
 | |
|         #   * DEVELOPMENT_LOG_EMAILS: Whether to log emails sent.
 | |
|         #
 | |
|         # We use our assertLogs() helper to catch log entries,
 | |
|         # as you'll see below. Read more about assertLogs() at:
 | |
|         # https://docs.python.org/3/library/unittest.html#unittest.TestCase.assertLogs
 | |
|         #
 | |
|         # We use mock.patch to simulate _do_send_messages.
 | |
|         with (
 | |
|             self.settings(EMAIL_BACKEND="zproject.email_backends.EmailLogBackEnd"),
 | |
|             self.settings(DEVELOPMENT_LOG_EMAILS=True),
 | |
|             self.assertLogs(level="INFO") as info_log,
 | |
|             mock.patch(
 | |
|                 "zproject.email_backends.EmailLogBackEnd._do_send_messages", lambda *args: 1
 | |
|             ),
 | |
|         ):
 | |
|             # Parts of this endpoint use transactions, and use
 | |
|             # transaction.on_commit to run code when the transaction
 | |
|             # commits.  Tests are run inside one big outer
 | |
|             # transaction, so those never get a chance to run unless
 | |
|             # we explicitly make a fake boundary to run them at; that
 | |
|             # is what captureOnCommitCallbacks does.
 | |
|             with self.captureOnCommitCallbacks(execute=True):
 | |
|                 result = self.client_get(
 | |
|                     "/emails/generate/"
 | |
|                 )  # Generates emails and redirects to /emails/
 | |
| 
 | |
|             # Verify redirect
 | |
|             self.assertEqual(result["Location"], "/emails/")
 | |
| 
 | |
|             # The above call to /emails/generate/ creates the emails and
 | |
|             # logs the below line for every email.
 | |
|             expected_log_line = (
 | |
|                 "INFO:root:Emails sent in development are available at http://testserver/emails"
 | |
|             )
 | |
| 
 | |
|             # info_log.output is a list of all the log messages captured.
 | |
|             self.assertEqual(info_log.output, [expected_log_line] * 20)
 | |
| 
 | |
|             # Now, lets actually go the URL the above call redirects to, i.e., /emails/
 | |
|             result = self.client_get(result["Location"])
 | |
| 
 | |
|             # assert_in_success_response() verifies that the content
 | |
|             # we received from client_get includes the strings we expect
 | |
|             self.assert_in_success_response(["All emails sent in the Zulip"], result)
 | |
| 
 | |
| 
 | |
| class TestMocking(ZulipTestCase):
 | |
|     # Mocking, primarily used in testing, is a technique that allows you to
 | |
|     # replace methods or objects with fake entities.
 | |
|     #
 | |
|     # Mocking is generally used in situations where
 | |
|     # we want to avoid running original code for reasons
 | |
|     # like skipping HTTP requests, saving execution time,
 | |
|     # or simulating convenient return values.
 | |
|     #
 | |
|     # Learn more about mocking in-depth at:
 | |
|     # https://zulip.readthedocs.io/en/latest/testing/testing-with-django.html#testing-with-mocks
 | |
|     def test_wildcard_mentions(self) -> None:
 | |
|         cordelia = self.example_user("cordelia")
 | |
|         realm = cordelia.realm
 | |
|         self.login_user(cordelia)
 | |
|         self.subscribe(cordelia, "test_stream")
 | |
| 
 | |
|         # Let's explicitly set the policy for sending wildcard
 | |
|         # mentions to a stream. If a stream has too many
 | |
|         # subscribers, we won't allow any users to spam the stream.
 | |
|         nobody_system_group = NamedUserGroup.objects.get(
 | |
|             name=SystemGroups.NOBODY, realm=realm, is_system_group=True
 | |
|         )
 | |
| 
 | |
|         do_change_realm_permission_group_setting(
 | |
|             realm,
 | |
|             "can_mention_many_users_group",
 | |
|             nobody_system_group,
 | |
|             acting_user=None,
 | |
|         )
 | |
| 
 | |
|         # We will try the same message a couple times.
 | |
|         # Notice the content includes "@**all**".
 | |
|         all_mention = "@**all** test wildcard mention"
 | |
| 
 | |
|         # First, let's test the SAD PATH, where cordelia
 | |
|         # tries a wildcard method and gets rejected.
 | |
|         #
 | |
|         # Here we use both mock.patch to simulate a return value
 | |
|         # and assertRaisesRegex to verify our function raises
 | |
|         # an error.
 | |
|         with (
 | |
|             mock.patch(
 | |
|                 "zerver.lib.message.num_subscribers_for_stream_id",
 | |
|                 return_value=Realm.WILDCARD_MENTION_THRESHOLD + 1,
 | |
|             ),
 | |
|             self.assertRaisesRegex(
 | |
|                 StreamWildcardMentionNotAllowedError,
 | |
|                 "You do not have permission to use channel wildcard mentions in this channel.",
 | |
|             ),
 | |
|         ):
 | |
|             self.send_stream_message(
 | |
|                 sender=cordelia,
 | |
|                 stream_name="test_stream",
 | |
|                 content=all_mention,
 | |
|             )
 | |
| 
 | |
|         # Verify the message was NOT sent.
 | |
|         message = most_recent_message(cordelia)
 | |
|         self.assertNotEqual(message.content, all_mention)
 | |
| 
 | |
|         # Now for the HAPPY PATH, we still mock the number of
 | |
|         # subscribers, but here we simulate that we are under
 | |
|         # the limit.  We expect cordelia's message to go through.
 | |
|         with mock.patch(
 | |
|             "zerver.lib.message.num_subscribers_for_stream_id",
 | |
|             return_value=Realm.WILDCARD_MENTION_THRESHOLD - 1,
 | |
|         ):
 | |
|             self.send_stream_message(
 | |
|                 sender=cordelia,
 | |
|                 stream_name="test_stream",
 | |
|                 content=all_mention,
 | |
|             )
 | |
| 
 | |
|         # Verify the message WAS sent.
 | |
|         message = most_recent_message(cordelia)
 | |
|         self.assertEqual(message.content, all_mention)
 | |
| 
 | |
| 
 | |
| class TestTimeTravel(ZulipTestCase):
 | |
|     def test_edit_message(self) -> None:
 | |
|         """
 | |
|         Verify if the time limit imposed on message editing is working correctly.
 | |
|         """
 | |
|         iago = self.example_user("iago")
 | |
|         self.login("iago")
 | |
| 
 | |
|         # Set limit to edit message content.
 | |
|         MESSAGE_CONTENT_EDIT_LIMIT = 5 * 60  # 5 minutes
 | |
|         result = self.client_patch(
 | |
|             "/json/realm",
 | |
|             {
 | |
|                 "allow_message_editing": "true",
 | |
|                 "message_content_edit_limit_seconds": MESSAGE_CONTENT_EDIT_LIMIT,
 | |
|             },
 | |
|         )
 | |
|         self.assert_json_success(result)
 | |
| 
 | |
|         sent_message_id = self.send_stream_message(
 | |
|             iago,
 | |
|             "Scotland",
 | |
|             topic_name="lunch",
 | |
|             content="I want pizza!",
 | |
|         )
 | |
|         message_sent_time = timezone_now()
 | |
| 
 | |
|         # Verify message sent.
 | |
|         message = most_recent_message(iago)
 | |
|         self.assertEqual(message.id, sent_message_id)
 | |
|         self.assertEqual(message.content, "I want pizza!")
 | |
| 
 | |
|         # Edit message content now. This should work as we're editing
 | |
|         # it immediately after sending i.e., before the limit exceeds.
 | |
|         result = self.client_patch(
 | |
|             f"/json/messages/{sent_message_id}", {"content": "I want burger!"}
 | |
|         )
 | |
|         self.assert_json_success(result)
 | |
|         message = most_recent_message(iago)
 | |
|         self.assertEqual(message.id, sent_message_id)
 | |
|         self.assertEqual(message.content, "I want burger!")
 | |
| 
 | |
|         # Now that we tested message editing works within the limit,
 | |
|         # we want to verify it doesn't work beyond the limit.
 | |
|         #
 | |
|         # First, calculate the time we want to travel to, adding
 | |
|         # a little buffer of 100 seconds beyond the limit.
 | |
|         time_beyond_edit_limit = message_sent_time + timedelta(
 | |
|             seconds=MESSAGE_CONTENT_EDIT_LIMIT + 100
 | |
|         )
 | |
| 
 | |
|         # Now use time_machine.travel to simulate that we are in the future.
 | |
|         #
 | |
|         # See https://pypi.org/project/time-machine/ for more info.
 | |
|         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."}
 | |
|             )
 | |
|             self.assert_json_error(result, msg="The time limit for editing this message has passed")
 | |
|             message = most_recent_message(iago)
 | |
|             self.assertEqual(message.id, sent_message_id)
 | |
|             self.assertEqual(message.content, "I want burger!")
 |