from collections.abc import Sequence
from datetime import datetime, timedelta, timezone

import time_machine
from django.conf import settings
from django.core import mail
from django.test import override_settings
from typing_extensions import override

from corporate.lib.stripe import get_latest_seat_count
from zerver.actions.create_user import notify_new_user
from zerver.actions.streams import do_set_stream_property
from zerver.actions.user_settings import do_change_user_setting
from zerver.lib.initial_password import initial_password
from zerver.lib.test_classes import ZulipTestCase
from zerver.models import Message, Recipient, Stream, UserProfile
from zerver.models.realms import get_realm
from zerver.models.recipients import get_direct_message_group_user_ids
from zerver.models.streams import StreamTopicsPolicyEnum, get_stream
from zerver.models.users import get_system_bot
from zerver.signals import JUST_CREATED_THRESHOLD, get_device_browser, get_device_os


class SendLoginEmailTest(ZulipTestCase):
    """
    Uses django's user_logged_in signal to send emails on new login.

    The receiver handler for this signal is always registered in production,
    development and testing, but emails are only sent based on SEND_LOGIN_EMAILS setting.

    SEND_LOGIN_EMAILS is set to true in default settings.
    It is turned off during testing.
    """

    def test_send_login_emails_if_send_login_email_setting_is_true(self) -> None:
        with self.settings(SEND_LOGIN_EMAILS=True):
            self.assertTrue(settings.SEND_LOGIN_EMAILS)
            # we don't use the self.login method since we spoof the user-agent
            mock_time = datetime(year=2018, month=1, day=1, tzinfo=timezone.utc)

            user = self.example_user("hamlet")
            user.timezone = "US/Pacific"
            user.twenty_four_hour_time = False
            user.date_joined = mock_time - timedelta(seconds=JUST_CREATED_THRESHOLD + 1)
            user.save()
            password = initial_password(user.delivery_email)
            login_info = dict(
                username=user.delivery_email,
                password=password,
            )
            firefox_windows = (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:47.0) Gecko/20100101 Firefox/47.0"
            )
            mock_time = datetime(year=2018, month=1, day=1, tzinfo=timezone.utc)
            with time_machine.travel(mock_time, tick=False):
                self.client_post(
                    "/accounts/login/", info=login_info, HTTP_USER_AGENT=firefox_windows
                )

            # email is sent and correct subject
            self.assert_length(mail.outbox, 1)
            subject = "New login from Firefox on Windows"
            self.assertEqual(mail.outbox[0].subject, subject)
            # local time is correct and in email's body
            self.assertRegex(str(mail.outbox[0].body), r"Sun, Dec 31, 2017, 4:00[ \u202f]PM PST")

            # Try again with the 24h time format setting enabled for this user
            self.logout()  # We just logged in, we'd be redirected without this
            user.twenty_four_hour_time = True
            user.save()
            with time_machine.travel(mock_time, tick=False):
                self.client_post(
                    "/accounts/login/", info=login_info, HTTP_USER_AGENT=firefox_windows
                )

            reference_time = "Sun, Dec 31, 2017, 16:00 PST"
            self.assertIn(reference_time, mail.outbox[1].body)

    def test_dont_send_login_emails_if_send_login_emails_is_false(self) -> None:
        self.assertFalse(settings.SEND_LOGIN_EMAILS)
        user = self.example_user("hamlet")
        self.login_user(user)

        self.assert_length(mail.outbox, 0)

    def test_dont_send_login_emails_for_new_user_registration_logins(self) -> None:
        with self.settings(SEND_LOGIN_EMAILS=True):
            self.register("test@zulip.com", "test")

            # Verify that there's just 1 email for new user registration.
            self.assertEqual(mail.outbox[0].subject, "Activate your Zulip account")
            self.assert_length(mail.outbox, 1)

    def test_without_path_info_dont_send_login_emails_for_new_user_registration_logins(
        self,
    ) -> None:
        with self.settings(SEND_LOGIN_EMAILS=True):
            self.client_post("/accounts/home/", {"email": "orange@zulip.com"})
            self.submit_reg_form_for_user("orange@zulip.com", "orange", PATH_INFO="")

            for email in mail.outbox:
                subject = "New login from an unknown browser on an unknown operating system"
                self.assertNotEqual(email.subject, subject)

    @override_settings(SEND_LOGIN_EMAILS=True)
    def test_enable_login_emails_user_setting(self) -> None:
        user = self.example_user("hamlet")
        mock_time = datetime(year=2018, month=1, day=1, tzinfo=timezone.utc)

        user.timezone = "US/Pacific"
        user.date_joined = mock_time - timedelta(seconds=JUST_CREATED_THRESHOLD + 1)
        user.save()

        do_change_user_setting(user, "enable_login_emails", False, acting_user=None)
        self.assertFalse(user.enable_login_emails)
        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 time_machine.travel(mock_time, tick=False):
            self.login_user(user)
        self.assert_length(mail.outbox, 1)


class TestBrowserAndOsUserAgentStrings(ZulipTestCase):
    @override
    def setUp(self) -> None:
        super().setUp()
        self.user_agents = [
            # Linux
            (
                (
                    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko)"
                    " Chrome/146.0.0.0 Safari/537.36"
                ),
                "Chrome",
                "Linux",
            ),
            (
                "Mozilla/5.0 (X11; Ubuntu; Linux i686; rv:149.0) Gecko/20100101 Firefox/149.0",
                "Firefox",
                "Ubuntu",
            ),
            (
                (
                    "Mozilla/5.0 (X11; Linux i686) AppleWebKit/535.7 (KHTML, like Gecko)"
                    " Ubuntu/11.10 Chromium/16.0.912.77 Chrome/16.0.912.77 Safari/535.7"
                ),
                "Chromium",
                "Ubuntu",
            ),
            # Windows
            (
                (
                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko)"
                    " Chrome/146.0.0.0 Safari/537.36"
                ),
                "Chrome",
                "Windows",
            ),
            (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:149.0) Gecko/20100101 Firefox/149.0",
                "Firefox",
                "Windows",
            ),
            (
                "Mozilla/5.0 (Windows NT 10.0; Trident/7.0; rv:11.0) like Gecko",
                "Internet Explorer",
                "Windows",
            ),
            (
                (
                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko)"
                    " Chrome/146.0.0.0 Safari/537.36 OPR/130.0.0.0"
                ),
                "Opera",
                "Windows",
            ),
            (
                (
                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko)"
                    " Chrome/146.0.0.0 Safari/537.36 Edg/147.0.3912.60"
                ),
                "Edge",
                "Windows",
            ),
            # Android
            (
                (
                    "Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko)"
                    " Chrome/146.0.7680.178 Mobile Safari/537.36"
                ),
                "Chrome Mobile",
                "Android",
            ),
            (
                "Mozilla/5.0 (Android 16; Mobile; rv:149.0) Gecko/149.0 Firefox/149.0",
                "Firefox Mobile",
                "Android",
            ),
            # iOS
            (
                (
                    "Mozilla/5.0 (iPhone; CPU iPhone OS 18_7 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko)"
                    " CriOS/147.0.7727.47 Mobile/15E148 Safari/604.1"
                ),
                "Chrome Mobile iOS",
                "iOS",
            ),
            (
                (
                    "Mozilla/5.0 (iPhone; CPU iPhone OS 15_7_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko)"
                    " FxiOS/149.0 Mobile/15E148 Safari/605.1.15"
                ),
                "Firefox iOS",
                "iOS",
            ),
            (
                (
                    "Mozilla/5.0 (iPhone; CPU iPhone OS 18_7_7 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko)"
                    " Version/26.0 Mobile/15E148 Safari/604.1"
                ),
                "Mobile Safari",
                "iOS",
            ),
            (
                (
                    "Mozilla/5.0 (iPhone; CPU iPhone OS 6_1_4 like Mac OS X)"
                    " AppleWebKit/536.26 (KHTML, like Gecko) Mobile/10B350"
                ),
                "Mobile Safari UI/WKWebView",
                "iOS",
            ),
            # iPad
            (
                (
                    "Mozilla/5.0 (iPad; CPU OS 18_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko)"
                    " Version/18.5 Mobile/15E148 Safari/604.1"
                ),
                "Mobile Safari",
                "iOS",
            ),
            # macOS
            (
                (
                    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko)"
                    " Chrome/146.0.0.0 Safari/537.36"
                ),
                "Chrome",
                "macOS",
            ),
            (
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 15.7; rv:149.0) Gecko/20100101 Firefox/149.0",
                "Firefox",
                "macOS",
            ),
            (
                (
                    "Mozilla/5.0 (Macintosh; Intel Mac OS X 15_7_5) AppleWebKit/605.1.15 (KHTML, like Gecko)"
                    " Version/26.0 Safari/605.1.15"
                ),
                "Safari",
                "macOS",
            ),
            # Miscellaneous
            ("ZulipAndroid/1.0", "Zulip", "Android"),
            ("ZulipMobile/1.0.12 (Android 7.1.1)", "Zulip", "Android"),
            ("ZulipMobile/0.7.1.1 (iOS 10.3.1)", "Zulip", "iOS"),
            (
                (
                    "ZulipElectron/1.1.0-beta Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
                    " AppleWebKit/537.36 (KHTML, like Gecko) Zulip/1.1.0-beta"
                    " Chrome/56.0.2924.87 Electron/1.6.8 Safari/537.36"
                ),
                "Zulip",
                "Windows",
            ),
            (
                (
                    "Mozilla/5.0 (X11; CrOS x86_64 16503.74.0) AppleWebKit/537.36 (KHTML, like Gecko)"
                    " Chrome/144.0.7559.172 Safari/537.36"
                ),
                "Chrome",
                "Chrome OS",
            ),
            ("", None, None),
        ]

    def test_get_browser_on_new_login(self) -> None:
        for user_agent in self.user_agents:
            device_browser = get_device_browser(user_agent[0])
            self.assertEqual(device_browser, user_agent[1])

    def test_get_os_on_new_login(self) -> None:
        for user_agent in self.user_agents:
            device_os = get_device_os(user_agent[0])
            self.assertEqual(device_os, user_agent[2])


class TestNotifyNewUser(ZulipTestCase):
    def get_message_count(self) -> int:
        return Message.objects.all().count()

    def test_notify_realm_of_new_user(self) -> None:
        realm = get_realm("zulip")
        realm.signup_announcements_stream = get_stream("core team", realm)
        realm.save(update_fields=["signup_announcements_stream"])
        new_user = self.example_user("cordelia")
        message_count = self.get_message_count()

        notify_new_user(new_user)
        self.assertEqual(self.get_message_count(), message_count + 1)
        message = self.get_last_message()
        self.assertEqual(message.recipient.type, Recipient.STREAM)
        actual_stream = Stream.objects.get(id=message.recipient.type_id)
        self.assertEqual(actual_stream.name, "core team")
        self.assertEqual(message.topic_name(), "signups")
        self.assertIn(
            f"@_**Cordelia, Lear's daughter|{new_user.id}** joined this organization.",
            message.content,
        )

        realm.signup_announcements_stream = None
        realm.save(update_fields=["signup_announcements_stream"])
        new_user.refresh_from_db()
        notify_new_user(new_user)
        self.assertEqual(self.get_message_count(), message_count + 1)

    def test_notify_realm_of_new_user_in_empty_topic_only_channel(self) -> None:
        realm = get_realm("zulip")
        iago = self.example_user("iago")
        stream = get_stream("core team", realm)
        realm.signup_announcements_stream = stream
        realm.save(update_fields=["signup_announcements_stream"])
        do_set_stream_property(
            stream, "topics_policy", StreamTopicsPolicyEnum.empty_topic_only.value, iago
        )

        new_user = self.example_user("cordelia")
        message_count = self.get_message_count()

        notify_new_user(new_user)
        self.assertEqual(self.get_message_count(), message_count + 1)
        message = self.get_last_message()
        self.assertEqual(message.topic_name(), "")
        self.assertIn(
            f"@_**Cordelia, Lear's daughter|{new_user.id}** joined this organization.",
            message.content,
        )

    def test_notify_realm_of_new_user_in_manual_license_management(self) -> None:
        realm = get_realm("zulip")
        admin_user_ids = set(realm.get_human_admin_users().values_list("id", flat=True))
        notification_bot = get_system_bot(settings.NOTIFICATION_BOT, realm.id)
        expected_group_direct_message_user_ids = admin_user_ids | {notification_bot.id}
        realm.signup_announcements_stream = get_stream("core team", realm)
        realm.save(update_fields=["signup_announcements_stream"])

        user_count = get_latest_seat_count(realm)
        extra_licenses = 5
        self.subscribe_realm_to_monthly_plan_on_manual_license_management(
            realm, user_count + extra_licenses, user_count + extra_licenses
        )

        user_no = 0

        def create_new_user_and_verify_strings_in_notification_message(
            strings_present: Sequence[str] = [], strings_absent: Sequence[str] = []
        ) -> None:
            nonlocal user_no
            user_no += 1
            new_user = UserProfile.objects.create(
                realm=realm,
                full_name=f"new user {user_no}",
                email=f"user-{user_no}-email@zulip.com",
                delivery_email=f"user-{user_no}-delivery-email@zulip.com",
            )
            notify_new_user(new_user)

            message = self.get_last_message()
            if extra_licenses - user_no > 3:
                # More than 3 licenses remaining. No group DM.
                actual_stream = Stream.objects.get(id=message.recipient.type_id)
                self.assertEqual(actual_stream, realm.signup_announcements_stream)
            else:
                # Stream message
                second_to_last_message = self.get_second_to_last_message()
                actual_stream = Stream.objects.get(id=second_to_last_message.recipient.type_id)
                self.assertEqual(actual_stream, realm.signup_announcements_stream)
                self.assertIn(
                    f"@_**new user {user_no}|{new_user.id}** joined this organization.",
                    second_to_last_message.content,
                )
                # Group DM
                self.assertEqual(
                    set(get_direct_message_group_user_ids(message.recipient)),
                    expected_group_direct_message_user_ids,
                )
            self.assertIn(
                f"@_**new user {user_no}|{new_user.id}** joined this organization.",
                message.content,
            )
            for string_present in strings_present:
                self.assertIn(
                    string_present,
                    message.content,
                )
            for string_absent in strings_absent:
                self.assertNotIn(string_absent, message.content)

        create_new_user_and_verify_strings_in_notification_message(
            strings_absent=["Your organization has"]
        )
        create_new_user_and_verify_strings_in_notification_message(
            strings_present=[
                "Your organization has only three Zulip licenses remaining",
                "to allow more than three users to",
            ],
        )
        create_new_user_and_verify_strings_in_notification_message(
            strings_present=[
                "Your organization has only two Zulip licenses remaining",
                "to allow more than two users to",
            ],
        )

        create_new_user_and_verify_strings_in_notification_message(
            strings_present=[
                "Your organization has only one Zulip license remaining",
                "to allow more than one user to",
            ],
        )
        create_new_user_and_verify_strings_in_notification_message(
            strings_present=[
                "Your organization has no Zulip licenses remaining",
                "to allow new users to",
            ]
        )