from datetime import timedelta
from typing import TYPE_CHECKING
from unittest import mock

import responses
import time_machine
from django.conf import settings
from django.utils.timezone import now as timezone_now
from typing_extensions import override

from corporate.lib.remote_billing_util import (
    REMOTE_BILLING_SESSION_VALIDITY_SECONDS,
    LegacyServerIdentityDict,
    RemoteBillingIdentityDict,
    RemoteBillingUserDict,
)
from corporate.lib.stripe import RemoteRealmBillingSession, RemoteServerBillingSession, add_months
from corporate.models.customers import get_customer_by_remote_realm, get_customer_by_remote_server
from corporate.models.licenses import LicenseLedger
from corporate.models.plans import CustomerPlan, get_current_plan_by_customer
from corporate.views.remote_billing_page import generate_confirmation_link_for_server_deactivation
from zerver.lib.exceptions import RemoteRealmServerMismatchError
from zerver.lib.rate_limiter import RateLimitedIPAddr
from zerver.lib.remote_server import send_server_data_to_push_bouncer
from zerver.lib.send_email import FromAddress
from zerver.lib.test_classes import BouncerTestCase
from zerver.lib.test_helpers import activate_push_notification_service, ratelimit_rule
from zerver.lib.timestamp import datetime_to_timestamp
from zerver.models import Realm, UserProfile
from zerver.models.realms import get_realm
from zilencer.models import (
    PreregistrationRemoteRealmBillingUser,
    PreregistrationRemoteServerBillingUser,
    RateLimitedRemoteZulipServer,
    RemoteRealm,
    RemoteRealmBillingUser,
    RemoteServerBillingUser,
    RemoteZulipServer,
)

if TYPE_CHECKING:
    from django.test.client import _MonkeyPatchedWSGIResponse as TestHttpResponse


class RemoteRealmBillingTestCase(BouncerTestCase):
    def execute_remote_billing_authentication_flow(
        self,
        user: UserProfile,
        next_page: str | None = None,
        expect_tos: bool = True,
        confirm_tos: bool = True,
        first_time_login: bool = True,
        # This only matters if first_time_login is True, since otherwise
        # there's no confirmation link to be clicked:
        return_without_clicking_confirmation_link: bool = False,
        # This is in order to return the response early, right after accessing the
        # authentication url for the user. This is useful for tests who expect an
        # an error there.
        return_from_auth_url: bool = False,
    ) -> "TestHttpResponse":
        now = timezone_now()

        self_hosted_billing_url = "/self-hosted-billing/"
        if next_page is not None:
            self_hosted_billing_url += f"?next_page={next_page}"
        with time_machine.travel(now, tick=False):
            result = self.client_get(self_hosted_billing_url)

        self.assertEqual(result.status_code, 302)
        self.assertIn("http://selfhosting.testserver/remote-billing-login/", result["Location"])

        signed_auth_url = result["Location"]
        signed_access_token = signed_auth_url.split("/")[-1]
        with time_machine.travel(now, tick=False):
            result = self.client_get(signed_auth_url, subdomain="selfhosting")

        if return_from_auth_url:
            return result

        if first_time_login:
            self.assertFalse(RemoteRealmBillingUser.objects.filter(user_uuid=user.uuid).exists())
            # When logging in for the first time some extra steps are needed
            # to confirm and verify the email address.
            self.assertEqual(result.status_code, 200)
            self.assert_in_success_response(["Enter email"], result)
            self.assert_in_success_response([user.realm.host], result)
            self.assert_in_success_response(
                [f'action="/remote-billing-login/{signed_access_token}/confirm/"'], result
            )

            with time_machine.travel(now, tick=False):
                result = self.client_post(
                    f"/remote-billing-login/{signed_access_token}/confirm/",
                    {"email": user.delivery_email},
                    subdomain="selfhosting",
                )
            if result.status_code == 429:
                # Return rate limit errors early, since they occur in rate limiting tests
                # that want to verify them.
                return result

            self.assertEqual(result.status_code, 200)
            self.assert_in_success_response(
                [
                    "To finish logging in, check your email account (",
                    ") for a confirmation email from Zulip.",
                    user.delivery_email,
                ],
                result,
            )
            confirmation_url = self.get_confirmation_url_from_outbox(
                user.delivery_email,
                url_pattern=(
                    f"{settings.SELF_HOSTING_MANAGEMENT_SUBDOMAIN}.{settings.EXTERNAL_HOST}"
                    r"(\S+)"
                ),
                email_body_contains="confirm your email and log in to Zulip plan management",
            )
            if return_without_clicking_confirmation_link:
                return result

            with time_machine.travel(now, tick=False):
                result = self.client_get(confirmation_url, subdomain="selfhosting")

            remote_billing_user = RemoteRealmBillingUser.objects.latest("id")
            self.assertEqual(remote_billing_user.user_uuid, user.uuid)
            self.assertEqual(remote_billing_user.email, user.delivery_email)

            prereg_user = PreregistrationRemoteRealmBillingUser.objects.latest("id")
            self.assertEqual(prereg_user.created_user, remote_billing_user)
            self.assertEqual(remote_billing_user.date_joined, now)

        # Final confirmation page - just confirm your details, possibly
        # agreeing to ToS if needed and an authenticated session will be granted:
        self.assertEqual(result.status_code, 200)
        self.assert_in_success_response(["Log in to Zulip plan management"], result)
        self.assert_in_success_response([user.realm.host], result)

        params = {}
        if expect_tos:
            self.assert_in_success_response(["I agree", "Terms of Service"], result)
        if confirm_tos:
            params = {"tos_consent": "true"}

        with time_machine.travel(now, tick=False):
            result = self.client_post(signed_auth_url, params, subdomain="selfhosting")
        if result.status_code >= 400:
            # Failures should be returned early so the caller can assert about them.
            return result

        # Verify the authed data that should have been stored in the session.
        remote_billing_user = RemoteRealmBillingUser.objects.get(user_uuid=user.uuid)
        identity_dict = RemoteBillingIdentityDict(
            user=RemoteBillingUserDict(
                user_email=user.delivery_email,
                user_uuid=str(user.uuid),
                user_full_name=user.full_name,
            ),
            remote_server_uuid=str(self.server.uuid),
            remote_realm_uuid=str(user.realm.uuid),
            remote_billing_user_id=remote_billing_user.id,
            authenticated_at=datetime_to_timestamp(now),
            uri_scheme="http://",
            next_page=next_page,
        )
        self.assertEqual(
            self.client.session["remote_billing_identities"][f"remote_realm:{user.realm.uuid!s}"],
            identity_dict,
        )

        self.assertEqual(remote_billing_user.last_login, now)

        # It's up to the caller to verify further details, such as the exact redirect URL,
        # depending on the set up and intent of the test.
        return result


@activate_push_notification_service()
class SelfHostedBillingEndpointBasicTest(RemoteRealmBillingTestCase):
    @responses.activate
    def test_self_hosted_billing_endpoints(self) -> None:
        # An ordinary user doesn't have access to these endpoints.
        self.login("hamlet")
        for url in [
            "/self-hosted-billing/",
            "/json/self-hosted-billing",
            "/self-hosted-billing/not-configured/",
        ]:
            result = self.client_get(url)
            self.assert_json_error(result, "Must be an organization owner")

        # Login as an organization owner to gain access.
        self.login("desdemona")

        self.add_mock_response()

        self_hosted_billing_url = "/self-hosted-billing/"
        self_hosted_billing_json_url = "/json/self-hosted-billing"

        with self.settings(ZULIP_SERVICE_PUSH_NOTIFICATIONS=False):
            with self.settings(CORPORATE_ENABLED=True):
                result = self.client_get(self_hosted_billing_url)
                self.assertEqual(result.status_code, 404)
                self.assert_in_response("Page not found (404)", result)

            with self.settings(CORPORATE_ENABLED=False):
                result = self.client_get(self_hosted_billing_url)
                self.assertEqual(result.status_code, 302)
                redirect_url = result["Location"]
                self.assertEqual(redirect_url, "/self-hosted-billing/not-configured/")

                with self.assertLogs("django.request"):
                    result = self.client_get(redirect_url)
                    self.assert_in_response(
                        "This server is not configured to use push notifications.", result
                    )

            with self.settings(CORPORATE_ENABLED=True):
                result = self.client_get(self_hosted_billing_json_url)
                self.assert_json_error(
                    result, "Server doesn't use the push notification service", 404
                )

            with self.settings(CORPORATE_ENABLED=False):
                result = self.client_get(self_hosted_billing_json_url)
                self.assert_json_success(result)

                redirect_url = result.json()["billing_access_url"]
                self.assertEqual(redirect_url, "/self-hosted-billing/not-configured/")

                with self.assertLogs("django.request"):
                    result = self.client_get(redirect_url)
                    self.assert_in_response(
                        "This server is not configured to use push notifications.", result
                    )

        with mock.patch(
            "zerver.views.push_notifications.send_to_push_bouncer",
            side_effect=RemoteRealmServerMismatchError,
        ):
            result = self.client_get(self_hosted_billing_url)
            self.assertEqual(result.status_code, 403)
            self.assert_in_response("Unexpected Zulip server registration", result)

            result = self.client_get(self_hosted_billing_json_url)
            self.assert_json_error(
                result,
                "Your organization is registered to a different Zulip server. Please contact Zulip support for assistance in resolving this issue.",
                403,
            )

        # Now test successes. We only check that an url for accessing the remote billing system
        # is returned (in the appropriate format - redirect or json data, depending on the endpoint).
        # We don't need to test that returned URL beyond that, because that's just the full auth flow,
        # which gets tested properly in other tests.
        result = self.client_get(self_hosted_billing_url)
        self.assertEqual(result.status_code, 302)
        self.assertIn("http://selfhosting.testserver/remote-billing-login/", result["Location"])

        result = self.client_get(self_hosted_billing_json_url)
        self.assert_json_success(result)
        data = result.json()
        self.assertEqual(sorted(data.keys()), ["billing_access_url", "msg", "result"])
        self.assertIn(
            "http://selfhosting.testserver/remote-billing-login/", data["billing_access_url"]
        )


@activate_push_notification_service()
class RemoteBillingAuthenticationTest(RemoteRealmBillingTestCase):
    def test_self_hosted_config_error_page(self) -> None:
        self.login("desdemona")

        with (
            self.settings(CORPORATE_ENABLED=False, ZULIP_SERVICE_PUSH_NOTIFICATIONS=False),
            self.assertLogs("django.request"),
        ):
            result = self.client_get("/self-hosted-billing/not-configured/")
            self.assertEqual(result.status_code, 500)
            self.assert_in_response(
                "This server is not configured to use push notifications.", result
            )

        # The page doesn't make sense if PUSH_NOTIFICATION_BOUNCER_URL is configured.
        with self.settings(CORPORATE_ENABLED=False):
            result = self.client_get("/self-hosted-billing/not-configured/")
            self.assertEqual(result.status_code, 404)

        # Also doesn't make sense on zulipchat.com (where CORPORATE_ENABLED is True).
        with self.settings(CORPORATE_ENABLED=True, ZULIP_SERVICE_PUSH_NOTIFICATIONS=False):
            result = self.client_get("/self-hosted-billing/not-configured/")
            self.assertEqual(result.status_code, 404)

    @responses.activate
    def test_remote_billing_authentication_flow(self) -> None:
        self.login("desdemona")
        desdemona = self.example_user("desdemona")
        realm = desdemona.realm

        self.add_mock_response()
        send_server_data_to_push_bouncer(consider_usage_statistics=False)

        result = self.execute_remote_billing_authentication_flow(desdemona)

        # TODO: The redirect URL will vary depending on the billing state of the user's
        # realm/server when we implement proper logic for that. For now, we can simply
        # hard-code an assert about the endpoint.
        self.assertEqual(result["Location"], f"/realm/{realm.uuid!s}/plans/")

        # Go to the URL we're redirected to after authentication and assert
        # some basic expected content.
        result = self.client_get(result["Location"], subdomain="selfhosting")
        self.assert_in_success_response(["showing-self-hosted", "Retain full control"], result)

    @ratelimit_rule(10, 3, domain="sends_email_by_remote_server")
    @ratelimit_rule(10, 2, domain="sends_email_by_ip")
    @responses.activate
    def test_remote_billing_authentication_flow_rate_limited(self) -> None:
        RateLimitedIPAddr("127.0.0.1", domain="sends_email_by_ip").clear_history()
        RateLimitedRemoteZulipServer(
            self.server, domain="sends_email_by_remote_server"
        ).clear_history()

        self.login("desdemona")
        desdemona = self.example_user("desdemona")

        self.add_mock_response()
        send_server_data_to_push_bouncer(consider_usage_statistics=False)

        for i in range(2):
            result = self.execute_remote_billing_authentication_flow(
                desdemona, return_without_clicking_confirmation_link=True
            )
            self.assertEqual(result.status_code, 200)

        result = self.execute_remote_billing_authentication_flow(
            desdemona, return_without_clicking_confirmation_link=True
        )
        self.assertEqual(result.status_code, 429)
        self.assert_in_response("You have exceeded the limit", result)

        # Reset the IP rate limit so that we trigger the server-based one.
        RateLimitedIPAddr("127.0.0.1", domain="sends_email_by_ip").clear_history()

        result = self.execute_remote_billing_authentication_flow(
            desdemona, return_without_clicking_confirmation_link=True
        )
        self.assertEqual(result.status_code, 200)

        with self.assertLogs("zilencer.auth", "WARN") as mock_log:
            result = self.execute_remote_billing_authentication_flow(
                desdemona, return_without_clicking_confirmation_link=True
            )
            self.assertEqual(result.status_code, 429)
            self.assert_in_response("Your server has exceeded the limit", result)
        self.assertEqual(
            mock_log.output,
            [
                f"WARNING:zilencer.auth:Remote server {self.server.hostname} {str(self.server.uuid)[:12]} exceeded "
                "rate limits on domain sends_email_by_remote_server"
            ],
        )

    @responses.activate
    def test_remote_billing_authentication_flow_realm_not_registered(self) -> None:
        RemoteRealm.objects.all().delete()

        self.login("desdemona")
        desdemona = self.example_user("desdemona")
        realm = desdemona.realm

        self.add_mock_response()

        # We do the flow without having the realm registered with the push bouncer.
        # In such a case, the local /self-hosted-billing/ endpoint should error-handle
        # properly and end up registering the server's realms with the bouncer,
        # and successfully completing the flow - transparently to the user.
        self.assertFalse(RemoteRealm.objects.filter(uuid=realm.uuid).exists())

        # send_server_data_to_push_bouncer will be called within the endpoint's
        # error handling to register realms with the bouncer. We mock.patch it
        # to be able to assert that it was called - but also use side_effect
        # to maintain the original behavior of the function, instead of
        # replacing it with a Mock.
        with mock.patch(
            "zerver.views.push_notifications.send_server_data_to_push_bouncer",
            side_effect=send_server_data_to_push_bouncer,
        ) as m:
            result = self.execute_remote_billing_authentication_flow(desdemona)

        m.assert_called_once()
        # The user's realm should now be registered:
        self.assertTrue(RemoteRealm.objects.filter(uuid=realm.uuid).exists())

        self.assertEqual(result["Location"], f"/realm/{realm.uuid!s}/plans/")

        result = self.client_get(result["Location"], subdomain="selfhosting")
        self.assert_in_success_response(["showing-self-hosted", "Retain full control"], result)

    @responses.activate
    def test_remote_billing_authentication_flow_tos_consent_failure(self) -> None:
        self.login("desdemona")
        desdemona = self.example_user("desdemona")

        self.add_mock_response()
        send_server_data_to_push_bouncer(consider_usage_statistics=False)

        result = self.execute_remote_billing_authentication_flow(
            desdemona,
            expect_tos=True,
            confirm_tos=False,
        )

        self.assert_json_error(result, "You must accept the Terms of Service to proceed.")

    @responses.activate
    def test_remote_billing_authentication_flow_tos_consent_update(self) -> None:
        self.login("desdemona")
        desdemona = self.example_user("desdemona")

        self.add_mock_response()
        send_server_data_to_push_bouncer(consider_usage_statistics=False)

        with self.settings(TERMS_OF_SERVICE_VERSION="1.0"):
            result = self.execute_remote_billing_authentication_flow(
                desdemona,
                expect_tos=True,
                confirm_tos=True,
            )

        self.assertEqual(result.status_code, 302)

        remote_billing_user = RemoteRealmBillingUser.objects.last()
        assert remote_billing_user is not None
        self.assertEqual(remote_billing_user.user_uuid, desdemona.uuid)
        self.assertEqual(remote_billing_user.tos_version, "1.0")

        # Now bump the ToS version. They need to agree again.
        with self.settings(TERMS_OF_SERVICE_VERSION="2.0"):
            result = self.execute_remote_billing_authentication_flow(
                desdemona,
                expect_tos=True,
                confirm_tos=False,
                first_time_login=False,
            )
            self.assert_json_error(result, "You must accept the Terms of Service to proceed.")

            result = self.execute_remote_billing_authentication_flow(
                desdemona,
                expect_tos=True,
                confirm_tos=True,
                first_time_login=False,
            )
        remote_billing_user.refresh_from_db()
        self.assertEqual(remote_billing_user.user_uuid, desdemona.uuid)
        self.assertEqual(remote_billing_user.tos_version, "2.0")

    @responses.activate
    def test_remote_billing_authentication_flow_expired_session(self) -> None:
        now = timezone_now()

        self.login("desdemona")
        desdemona = self.example_user("desdemona")
        realm = desdemona.realm

        self.add_mock_response()
        send_server_data_to_push_bouncer(consider_usage_statistics=False)

        with time_machine.travel(now, tick=False):
            result = self.execute_remote_billing_authentication_flow(desdemona)

        self.assertEqual(result["Location"], f"/realm/{realm.uuid!s}/plans/")

        final_url = result["Location"]

        # Go to the URL we're redirected to after authentication and make sure
        # we're granted access.
        with time_machine.travel(
            now + timedelta(seconds=1),
            tick=False,
        ):
            result = self.client_get(final_url, subdomain="selfhosting")
        self.assert_in_success_response(["showing-self-hosted", "Retain full control"], result)

        # Now go there again, simulating doing this after the session has expired.
        # We should be denied access and redirected to re-auth.
        with time_machine.travel(
            now + timedelta(seconds=REMOTE_BILLING_SESSION_VALIDITY_SECONDS + 1),
            tick=False,
        ):
            result = self.client_get(
                final_url, subdomain="selfhosting", HTTP_ACCEPT="text/html, */*;q=0.8"
            )

            self.assertEqual(result.status_code, 302)
            self.assertEqual(
                result["Location"],
                f"http://{desdemona.realm.host}/self-hosted-billing/?next_page=plans",
            )

            # Opening this re-auth URL in result["Location"] is same as re-doing the auth
            # flow via execute_remote_billing_authentication_flow with next_page="plans".
            # So let's test that and assert that we end up successfully re-authed on the /plans
            # page.
            result = self.execute_remote_billing_authentication_flow(
                desdemona,
                next_page="plans",
                # ToS has already been confirmed earlier.
                expect_tos=False,
                confirm_tos=False,
                first_time_login=False,
            )
            self.assertEqual(result["Location"], f"/realm/{realm.uuid!s}/plans/")
            result = self.client_get(result["Location"], subdomain="selfhosting")
            self.assert_in_success_response(["showing-self-hosted", "Retain full control"], result)

    @responses.activate
    def test_remote_billing_unauthed_access(self) -> None:
        now = timezone_now()
        self.login("desdemona")
        desdemona = self.example_user("desdemona")
        realm = desdemona.realm

        self.add_mock_response()
        send_server_data_to_push_bouncer(consider_usage_statistics=False)

        # Straight-up access without authing at all:
        result = self.client_get(f"/realm/{realm.uuid!s}/plans/", subdomain="selfhosting")
        self.assert_json_error(result, "User not authenticated", 401)

        result = self.execute_remote_billing_authentication_flow(desdemona)
        self.assertEqual(result["Location"], f"/realm/{realm.uuid!s}/plans/")

        final_url = result["Location"]

        # Sanity check - access is granted after authing:
        result = self.client_get(final_url, subdomain="selfhosting")
        self.assertEqual(result.status_code, 200)

        # Now mess with the identity dict in the session in unlikely ways so that it should
        # not grant access.
        # First delete the RemoteRealm entry for this session.
        RemoteRealm.objects.filter(uuid=realm.uuid).delete()

        with self.assertLogs("django.request", "ERROR") as m, self.assertRaises(AssertionError):
            self.client_get(final_url, subdomain="selfhosting")
        self.assertIn(
            "The remote realm is missing despite being in the RemoteBillingIdentityDict",
            m.output[0],
        )

        # Try the case where the identity dict is simultaneously expired.
        with (
            time_machine.travel(
                now + timedelta(seconds=REMOTE_BILLING_SESSION_VALIDITY_SECONDS + 30),
                tick=False,
            ),
            self.assertLogs("django.request", "ERROR") as m,
            self.assertRaises(AssertionError),
        ):
            self.client_get(final_url, subdomain="selfhosting")
        # The django.request log should be a traceback, mentioning the relevant
        # exceptions that occurred.
        self.assertIn(
            "RemoteBillingIdentityExpiredError",
            m.output[0],
        )
        self.assertIn(
            "AssertionError",
            m.output[0],
        )

    @responses.activate
    def test_remote_billing_authentication_flow_to_sponsorship_page(self) -> None:
        self.login("desdemona")
        desdemona = self.example_user("desdemona")
        realm = desdemona.realm

        self.add_mock_response()
        send_server_data_to_push_bouncer(consider_usage_statistics=False)

        result = self.execute_remote_billing_authentication_flow(desdemona, "sponsorship")

        self.assertEqual(result["Location"], f"/realm/{realm.uuid!s}/sponsorship/")

        # Go to the URL we're redirected to after authentication and assert
        # some basic expected content.
        result = self.client_get(result["Location"], subdomain="selfhosting")
        self.assert_in_success_response(
            ["Request Zulip", "sponsorship", "Description of your organization"], result
        )

    @responses.activate
    def test_remote_billing_authentication_flow_to_upgrade_page(self) -> None:
        self.login("desdemona")
        desdemona = self.example_user("desdemona")
        realm = desdemona.realm

        self.add_mock_response()
        send_server_data_to_push_bouncer(consider_usage_statistics=False)

        result = self.execute_remote_billing_authentication_flow(desdemona, "upgrade")

        self.assertEqual(result["Location"], f"/realm/{realm.uuid!s}/upgrade/")

        # Go to the URL we're redirected to after authentication and assert
        # some basic expected content.
        # TODO: Add test for the case when redirected to error page (not yet implemented)
        # due to MissingDataError ('has_stale_audit_log' is True).
        with mock.patch("corporate.lib.stripe.has_stale_audit_log", return_value=False):
            result = self.client_get(result["Location"], subdomain="selfhosting")
            self.assert_in_success_response(
                ["Upgrade", "Purchase Zulip", "Your subscription will renew automatically."], result
            )

    @responses.activate
    def test_remote_billing_authentication_flow_cant_access_billing_without_finishing_confirmation(
        self,
    ) -> None:
        self.login("desdemona")
        desdemona = self.example_user("desdemona")
        realm = desdemona.realm

        self.add_mock_response()

        result = self.execute_remote_billing_authentication_flow(
            desdemona,
            expect_tos=True,
            confirm_tos=False,
            first_time_login=True,
            return_without_clicking_confirmation_link=True,
        )
        result = self.client_get(f"/realm/{realm.uuid!s}/billing/", subdomain="selfhosting")
        # Access is not allowed. The user doesn't have an IdentityDict in the session, so
        # we can't do a nice redirect back to their original server.
        self.assertEqual(result.status_code, 401)

    @responses.activate
    def test_remote_billing_authentication_flow_generate_two_confirmation_links_before_confirming(
        self,
    ) -> None:
        self.login("desdemona")
        desdemona = self.example_user("desdemona")

        self.add_mock_response()

        result = self.execute_remote_billing_authentication_flow(
            desdemona,
            expect_tos=True,
            confirm_tos=False,
            first_time_login=True,
            return_without_clicking_confirmation_link=True,
        )
        self.assertEqual(result.status_code, 200)
        first_confirmation_url = self.get_confirmation_url_from_outbox(
            desdemona.delivery_email,
            url_pattern=(
                f"{settings.SELF_HOSTING_MANAGEMENT_SUBDOMAIN}.{settings.EXTERNAL_HOST}" + r"(\S+)"
            ),
        )
        first_prereg_user = PreregistrationRemoteRealmBillingUser.objects.latest("id")

        result = self.execute_remote_billing_authentication_flow(
            desdemona,
            expect_tos=True,
            confirm_tos=False,
            first_time_login=True,
            return_without_clicking_confirmation_link=True,
        )
        self.assertEqual(result.status_code, 200)
        second_confirmation_url = self.get_confirmation_url_from_outbox(
            desdemona.delivery_email,
            url_pattern=(
                f"{settings.SELF_HOSTING_MANAGEMENT_SUBDOMAIN}.{settings.EXTERNAL_HOST}" + r"(\S+)"
            ),
        )
        second_prereg_user = PreregistrationRemoteRealmBillingUser.objects.latest("id")

        self.assertNotEqual(first_confirmation_url, second_confirmation_url)
        self.assertNotEqual(first_prereg_user.id, second_prereg_user.id)

        now = timezone_now()
        # Click the first confirmation link.
        with time_machine.travel(now, tick=False):
            result = self.client_get(first_confirmation_url, subdomain="selfhosting")
        self.assertEqual(result.status_code, 200)
        self.assert_in_success_response(["Log in to Zulip plan management"], result)

        # This created the RemoteRealmBillingUser entry.
        remote_billing_user = RemoteRealmBillingUser.objects.latest("id")
        self.assertEqual(remote_billing_user.user_uuid, desdemona.uuid)
        self.assertEqual(remote_billing_user.email, desdemona.delivery_email)

        first_prereg_user.refresh_from_db()
        self.assertEqual(first_prereg_user.created_user, remote_billing_user)

        # Now click the second confirmation link. The RemoteRealmBillingUser entry
        # stays the same, since it's already been created, and the user proceeds
        # normally further through the flow, while we log this event.
        with (
            time_machine.travel(now + timedelta(seconds=1), tick=False),
            self.assertLogs("corporate.stripe", "INFO") as mock_logger,
        ):
            result = self.client_get(second_confirmation_url, subdomain="selfhosting")
        self.assertEqual(result.status_code, 200)
        self.assert_in_success_response(["Log in to Zulip plan management"], result)

        # The RemoteRealmBillingUser entry stays the same.
        self.assertEqual(RemoteRealmBillingUser.objects.latest("id"), remote_billing_user)
        # The second prereg user is unused, since it wasn't needed.
        self.assertEqual(second_prereg_user.created_user, None)

        self.assertEqual(
            mock_logger.output,
            [
                "INFO:corporate.stripe:Matching RemoteRealmBillingUser already exists for "
                f"PreregistrationRemoteRealmBillingUser {second_prereg_user.id}"
            ],
        )

    @responses.activate
    def test_transfer_complimentary_access_plan_scheduled_for_upgrade_from_server_to_realm(
        self,
    ) -> None:
        self.login("desdemona")
        desdemona = self.example_user("desdemona")

        # Assert current server is not on any plan.
        self.assertIsNone(get_customer_by_remote_server(self.server))

        start_date = timezone_now()
        end_date = add_months(timezone_now(), 10)

        # Migrate server to complimentary access plan.
        server_billing_session = RemoteServerBillingSession(self.server)
        server_billing_session.create_complimentary_access_plan(start_date, end_date)

        server_customer = server_billing_session.get_customer()
        assert server_customer is not None
        server_plan = get_current_plan_by_customer(server_customer)
        assert server_plan is not None
        self.assertEqual(self.server.plan_type, RemoteZulipServer.PLAN_TYPE_SELF_MANAGED_LEGACY)
        self.assertEqual(server_plan.tier, CustomerPlan.TIER_SELF_HOSTED_LEGACY)
        self.assertEqual(server_plan.status, CustomerPlan.ACTIVE)

        # Schedule upgrade for plan.
        server_plan.status = CustomerPlan.SWITCH_PLAN_TIER_AT_PLAN_END
        server_plan.save(update_fields=["status"])

        # Just create a temporary plan and check if gets transferred or not.
