<commit_before>from typing import Any, List, Mapping

from django.db import connection

from zerver.models import (
    get_realm,
    get_stream,
    get_user,
    Subscription,
    UserMessage,
    UserProfile,
)

from zerver.lib.fix_unreads import (
    fix,
    fix_pre_pointer,
    fix_unsubscribed,
)
from zerver.lib.test_helpers import (
    get_subscription,
    tornado_redirected_to_list,
)
from zerver.lib.test_classes import (
    ZulipTestCase,
)
from zerver.lib.topic_mutes import add_topic_mute

import mock
import ujson

class PointerTest(ZulipTestCase):

    def test_update_pointer(self) -> None:
        """
        Posting a pointer to /update (in the form {"pointer": pointer}) changes
        the pointer we store for your UserProfile.
        """
        self.login('hamlet')
        self.assertEqual(self.example_user('hamlet').pointer, -1)
        msg_id = self.send_stream_message(self.example_user("othello"), "Verona")
        result = self.client_post("/json/users/me/pointer", {"pointer": msg_id})
        self.assert_json_success(result)
        self.assertEqual(self.example_user('hamlet').pointer, msg_id)

    def test_api_update_pointer(self) -> None:
        """
        Same as above, but for the API view
        """
        user = self.example_user('hamlet')
        email = user.email
        self.assertEqual(user.pointer, -1)
        msg_id = self.send_stream_message(self.example_user("othello"), "Verona")
        result = self.api_post(user, "/api/v1/users/me/pointer", {"pointer": msg_id})
        self.assert_json_success(result)
        self.assertEqual(get_user(email, user.realm).pointer, msg_id)

    def test_missing_pointer(self) -> None:
        """
        Posting json to /json/users/me/pointer which does not contain a pointer key/value pair
        returns a 400 and error message.
        """
        self.login('hamlet')
        self.assertEqual(self.example_user('hamlet').pointer, -1)
        result = self.client_post("/json/users/me/pointer", {"foo": 1})
        self.assert_json_error(result, "Missing 'pointer' argument")
        self.assertEqual(self.example_user('hamlet').pointer, -1)

    def test_invalid_pointer(self) -> None:
        """
        Posting json to /json/users/me/pointer with an invalid pointer returns a 400 and error
        message.
        """
        self.login('hamlet')
        self.assertEqual(self.example_user('hamlet').pointer, -1)
        result = self.client_post("/json/users/me/pointer", {"pointer": "foo"})
        self.assert_json_error(result, "Bad value for 'pointer': foo")
        self.assertEqual(self.example_user('hamlet').pointer, -1)

    def test_pointer_out_of_range(self) -> None:
        """
        Posting json to /json/users/me/pointer with an out of range (< 0) pointer returns a 400
        and error message.
        """
        self.login('hamlet')
        self.assertEqual(self.example_user('hamlet').pointer, -1)
        result = self.client_post("/json/users/me/pointer", {"pointer": -2})
        self.assert_json_error(result, "Bad value for 'pointer': -2")
        self.assertEqual(self.example_user('hamlet').pointer, -1)

    def test_use_first_unread_anchor_interaction_with_pointer(self) -> None:
        """
        Getting old messages (a get request to /json/messages) should never
        return an unread message older than the current pointer, when there's
        no narrow set.
        """
        self.login('hamlet')
        # Ensure the pointer is not set (-1)
        self.assertEqual(self.example_user('hamlet').pointer, -1)

        # Mark all existing messages as read
        result = self.client_post("/json/mark_all_as_read")
        self.assert_json_success(result)

        # Send a new message (this will be unread)
        new_message_id = self.send_stream_message(self.example_user("othello"), "Verona",
                                                  "test")

        # If we call get_messages with use_first_unread_anchor=True, we
        # should get the message we just sent
        messages_response = self.get_messages_response(
            anchor="first_unread", num_before=0, num_after=1)
        self.assertEqual(messages_response['messages'][0]['id'], new_message_id)
        self.assertEqual(messages_response['anchor'], new_message_id)

        # Test with the old way of expressing use_first_unread_anchor=True
        messages_response = self.get_messages_response(
            anchor=0, num_before=0, num_after=1, use_first_unread_anchor=True)
        self.assertEqual(messages_response['messages'][0]['id'], new_message_id)
        self.assertEqual(messages_response['anchor'], new_message_id)

        # We want to get the message_id of an arbitrary old message. We can
        # call get_messages with use_first_unread_anchor=False and simply
        # save the first message we're returned.
        messages = self.get_messages(
            anchor=0, num_before=0, num_after=2, use_first_unread_anchor=False)
        old_message_id = messages[0]['id']
        next_old_message_id = messages[1]['id']

        # Verify the message is marked as read
        user_message = UserMessage.objects.get(
            message_id=old_message_id,
            user_profile=self.example_user('hamlet'))
        self.assertTrue(user_message.flags.read)

        # Let's set this old message to be unread
        result = self.client_post("/json/messages/flags",
                                  {"messages": ujson.dumps([old_message_id]),
                                   "op": "remove",
                                   "flag": "read"})

        # Verify it's now marked as unread
        user_message = UserMessage.objects.get(
            message_id=old_message_id,
            user_profile=self.example_user('hamlet'))
        self.assert_json_success(result)
        self.assertFalse(user_message.flags.read)

        # Now if we call get_messages with use_first_unread_anchor=True,
        # we should get the old message we just set to unread
        messages_response = self.get_messages_response(
            anchor="first_unread", num_before=0, num_after=1)
        self.assertEqual(messages_response['messages'][0]['id'], old_message_id)
        self.assertEqual(messages_response['anchor'], old_message_id)

        # Let's update the pointer to be *after* this old unread message (but
        # still on or before the new unread message we just sent)
        result = self.client_post("/json/users/me/pointer",
                                  {"pointer": next_old_message_id})
        self.assert_json_success(result)
        self.assertEqual(self.example_user('hamlet').pointer,
                         next_old_message_id)

        # Verify that moving the pointer didn't mark our message as read.
        user_message = UserMessage.objects.get(
            message_id=old_message_id,
            user_profile=self.example_user('hamlet'))
        self.assertFalse(user_message.flags.read)

        # Now if we call get_messages with use_first_unread_anchor=True,
        # we should not get the old unread message (because it's before the
        # pointer), and instead should get the newly sent unread message
        messages_response = self.get_messages_response(
            anchor="first_unread", num_before=0, num_after=1)
        self.assertEqual(messages_response['messages'][0]['id'], new_message_id)
        self.assertEqual(messages_response['anchor'], new_message_id)

    def test_visible_messages_use_first_unread_anchor(self) -> None:
        self.login('hamlet')
        self.assertEqual(self.example_user('hamlet').pointer, -1)

        result = self.client_post("/json/mark_all_as_read")
        self.assert_json_success(result)

        new_message_id = self.send_stream_message(self.example_user("othello"), "Verona",
                                                  "test")

        messages_response = self.get_messages_response(
            anchor="first_unread", num_before=0, num_after=1)
        self.assertEqual(messages_response['messages'][0]['id'], new_message_id)
        self.assertEqual(messages_response['anchor'], new_message_id)

        with mock.patch('zerver.views.messages.get_first_visible_message_id', return_value=new_message_id):
            messages_response = self.get_messages_response(
                anchor="first_unread", num_before=0, num_after=1)
        self.assertEqual(messages_response['messages'][0]['id'], new_message_id)
        self.assertEqual(messages_response['anchor'], new_message_id)

        with mock.patch('zerver.views.messages.get_first_visible_message_id', return_value=new_message_id + 1):
            messages_reponse = self.get_messages_response(
                anchor="first_unread", num_before=0, num_after=1)
        self.assert_length(messages_reponse['messages'], 0)
        self.assertIn('anchor', messages_reponse)

        with mock.patch('zerver.views.messages.get_first_visible_message_id', return_value=new_message_id - 1):
            messages = self.get_messages(<commit_msg>test_unread: Remove test for removed mobile hack.<commit_after>from typing import Any, List, Mapping

from django.db import connection

from zerver.models import (
    get_realm,
    get_stream,
    get_user,
    Subscription,
    UserMessage,
    UserProfile,
)

from zerver.lib.fix_unreads import (
    fix,
    fix_pre_pointer,
    fix_unsubscribed,
)
from zerver.lib.test_helpers import (
    get_subscription,
    tornado_redirected_to_list,
)
from zerver.lib.test_classes import (
    ZulipTestCase,
)
from zerver.lib.topic_mutes import add_topic_mute

import mock
import ujson

class PointerTest(ZulipTestCase):

    def test_update_pointer(self) -> None:
        """
        Posting a pointer to /update (in the form {"pointer": pointer}) changes
        the pointer we store for your UserProfile.
        """
        self.login('hamlet')
        self.assertEqual(self.example_user('hamlet').pointer, -1)
        msg_id = self.send_stream_message(self.example_user("othello"), "Verona")
        result = self.client_post("/json/users/me/pointer", {"pointer": msg_id})
        self.assert_json_success(result)
        self.assertEqual(self.example_user('hamlet').pointer, msg_id)

    def test_api_update_pointer(self) -> None:
        """
        Same as above, but for the API view
        """
        user = self.example_user('hamlet')
        email = user.email
        self.assertEqual(user.pointer, -1)
        msg_id = self.send_stream_message(self.example_user("othello"), "Verona")
        result = self.api_post(user, "/api/v1/users/me/pointer", {"pointer": msg_id})
        self.assert_json_success(result)
        self.assertEqual(get_user(email, user.realm).pointer, msg_id)

    def test_missing_pointer(self) -> None:
        """
        Posting json to /json/users/me/pointer which does not contain a pointer key/value pair
        returns a 400 and error message.
        """
        self.login('hamlet')
        self.assertEqual(self.example_user('hamlet').pointer, -1)
        result = self.client_post("/json/users/me/pointer", {"foo": 1})
        self.assert_json_error(result, "Missing 'pointer' argument")
        self.assertEqual(self.example_user('hamlet').pointer, -1)

    def test_invalid_pointer(self) -> None:
        """
        Posting json to /json/users/me/pointer with an invalid pointer returns a 400 and error
        message.
        """
        self.login('hamlet')
        self.assertEqual(self.example_user('hamlet').pointer, -1)
        result = self.client_post("/json/users/me/pointer", {"pointer": "foo"})
        self.assert_json_error(result, "Bad value for 'pointer': foo")
        self.assertEqual(self.example_user('hamlet').pointer, -1)

    def test_pointer_out_of_range(self) -> None:
        """
        Posting json to /json/users/me/pointer with an out of range (< 0) pointer returns a 400
        and error message.
        """
        self.login('hamlet')
        self.assertEqual(self.example_user('hamlet').pointer, -1)
        result = self.client_post("/json/users/me/pointer", {"pointer": -2})
        self.assert_json_error(result, "Bad value for 'pointer': -2")
        self.assertEqual(self.example_user('hamlet').pointer, -1)

    def test_use_first_unread_anchor_interaction_with_pointer(self) -> None:
        """
        Getting old messages (a get request to /json/messages) should never
        return an unread message older than the current pointer, when there's
        no narrow set.
        """
        self.login('hamlet')
        # Ensure the pointer is not set (-1)
        self.assertEqual(self.example_user('hamlet').pointer, -1)

        # Mark all existing messages as read
        result = self.client_post("/json/mark_all_as_read")
        self.assert_json_success(result)

        # Send a new message (this will be unread)
        new_message_id = self.send_stream_message(self.example_user("othello"), "Verona",
                                                  "test")

        # If we call get_messages with use_first_unread_anchor=True, we
        # should get the message we just sent
        messages_response = self.get_messages_response(
            anchor="first_unread", num_before=0, num_after=1)
        self.assertEqual(messages_response['messages'][0]['id'], new_message_id)
        self.assertEqual(messages_response['anchor'], new_message_id)

        # Test with the old way of expressing use_first_unread_anchor=True
        messages_response = self.get_messages_response(
            anchor=0, num_before=0, num_after=1, use_first_unread_anchor=True)
        self.assertEqual(messages_response['messages'][0]['id'], new_message_id)
        self.assertEqual(messages_response['anchor'], new_message_id)

        # We want to get the message_id of an arbitrary old message. We can
        # call get_messages with use_first_unread_anchor=False and simply
        # save the first message we're returned.
        messages = self.get_messages(
            anchor=0, num_before=0, num_after=2, use_first_unread_anchor=False)
        old_message_id = messages[0]['id']
        next_old_message_id = messages[1]['id']

        # Verify the message is marked as read
        user_message = UserMessage.objects.get(
            message_id=old_message_id,
            user_profile=self.example_user('hamlet'))
        self.assertTrue(user_message.flags.read)

        # Let's set this old message to be unread
        result = self.client_post("/json/messages/flags",
                                  {"messages": ujson.dumps([old_message_id]),
                                   "op": "remove",
                                   "flag": "read"})

        # Verify it's now marked as unread
        user_message = UserMessage.objects.get(
            message_id=old_message_id,
            user_profile=self.example_user('hamlet'))
        self.assert_json_success(result)
        self.assertFalse(user_message.flags.read)

        # Now if we call get_messages with use_first_unread_anchor=True,
        # we should get the old message we just set to unread
        messages_response = self.get_messages_response(
            anchor="first_unread", num_before=0, num_after=1)
        self.assertEqual(messages_response['messages'][0]['id'], old_message_id)
        self.assertEqual(messages_response['anchor'], old_message_id)

        # Let's update the pointer to be *after* this old unread message (but
        # still on or before the new unread message we just sent)
        result = self.client_post("/json/users/me/pointer",
                                  {"pointer": next_old_message_id})
        self.assert_json_success(result)
        self.assertEqual(self.example_user('hamlet').pointer,
                         next_old_message_id)

        # Verify that moving the pointer didn't mark our message as read.
        user_message = UserMessage.objects.get(
            message_id=old_message_id,
            user_profile=self.example_user('hamlet'))
        self.assertFalse(user_message.flags.read)

    def test_visible_messages_use_first_unread_anchor(self) -> None:
        self.login('hamlet')
        self.assertEqual(self.example_user('hamlet').pointer, -1)

        result = self.client_post("/json/mark_all_as_read")
        self.assert_json_success(result)

        new_message_id = self.send_stream_message(self.example_user("othello"), "Verona",
                                                  "test")

        messages_response = self.get_messages_response(
            anchor="first_unread", num_before=0, num_after=1)
        self.assertEqual(messages_response['messages'][0]['id'], new_message_id)
        self.assertEqual(messages_response['anchor'], new_message_id)

        with mock.patch('zerver.views.messages.get_first_visible_message_id', return_value=new_message_id):
            messages_response = self.get_messages_response(
                anchor="first_unread", num_before=0, num_after=1)
        self.assertEqual(messages_response['messages'][0]['id'], new_message_id)
        self.assertEqual(messages_response['anchor'], new_message_id)

        with mock.patch('zerver.views.messages.get_first_visible_message_id', return_value=new_message_id + 1):
            messages_reponse = self.get_messages_response(
                anchor="first_unread", num_before=0, num_after=1)
        self.assert_length(messages_reponse['messages'], 0)
        self.assertIn('anchor', messages_reponse)

        with mock.patch('zerver.views.messages.get_first_visible_message_id', return_value=new_message_id - 1):
            messages = self.get_messages(
{'fraction_non_alphanumeric': 0.10584768812330009, 'fraction_numerical': 0.005949682683590208, 'mean_word_length': 2.815175097276265, 'pattern_counts': {'":': 20, '<': 5, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2946206', 'n_tokens_mistral': 5127, 'n_tokens_neox': 5028, 'n_words': 1059}