import os
import re
import tempfile
from datetime import timedelta
from io import StringIO
from unittest import mock
from unittest.mock import patch
from urllib.parse import quote

import orjson
import pyvips
import time_machine
from django.conf import settings
from django.core.files.uploadedfile import SimpleUploadedFile
from django.utils.timezone import now as timezone_now
from pyvips import at_least_libvips
from pyvips import version as libvips_version
from typing_extensions import override

import zerver.lib.upload
from analytics.models import RealmCount
from zerver.actions.create_realm import do_create_realm
from zerver.actions.create_user import do_create_user
from zerver.actions.message_send import internal_send_private_message
from zerver.actions.realm_icon import do_change_icon_source
from zerver.actions.realm_logo import do_change_logo_source
from zerver.actions.realm_settings import do_change_realm_plan_type, do_set_realm_property
from zerver.actions.user_settings import do_scrub_avatar_image
from zerver.lib.attachments import validate_attachment_request
from zerver.lib.avatar import (
    DEFAULT_AVATAR_FILE,
    avatar_url,
    get_avatar_field,
    get_static_avatar_url,
)
from zerver.lib.cache import cache_delete, cache_get, get_realm_used_upload_space_cache_key
from zerver.lib.create_user import copy_default_settings
from zerver.lib.initial_password import initial_password
from zerver.lib.realm_icon import realm_icon_url
from zerver.lib.realm_logo import get_realm_logo_url
from zerver.lib.test_classes import UploadSerializeMixin, ZulipTestCase
from zerver.lib.test_helpers import (
    avatar_disk_path,
    consume_response,
    get_test_image_file,
    ratelimit_rule,
)
from zerver.lib.upload import sanitize_name, upload_message_attachment
from zerver.lib.upload.base import ZulipUploadBackend
from zerver.lib.upload.local import LocalUploadBackend
from zerver.lib.upload.s3 import S3UploadBackend
from zerver.models import Attachment, Message, OnboardingStep, Realm, RealmDomain, UserProfile
from zerver.models.realms import get_realm
from zerver.models.users import get_system_bot, get_user_by_delivery_email


class FileUploadTest(UploadSerializeMixin, ZulipTestCase):
    def test_rest_endpoint(self) -> None:
        """
        Tests the /api/v1/user_uploads API endpoint. Here a single file is uploaded
        and downloaded using a username and api_key
        """
        fp = StringIO("zulip!")
        fp.name = "zulip.txt"

        # Upload file via API
        result = self.api_post(self.example_user("hamlet"), "/api/v1/user_uploads", {"file": fp})
        response_dict = self.assert_json_success(result)
        self.assertIn("uri", response_dict)
        self.assertIn("url", response_dict)
        url = response_dict["url"]
        self.assertEqual(response_dict["uri"], url)
        base = "/user_uploads/"
        self.assertEqual(base, url[: len(base)])

        # Download file via API
        self.logout()
        response = self.api_get(self.example_user("hamlet"), url)
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.getvalue(), b"zulip!")

        # Files uploaded through the API should be accessible via the web client
        self.login("hamlet")
        self.assertEqual(self.client_get(url).getvalue(), b"zulip!")

    def test_mobile_api_endpoint(self) -> None:
        """
        Tests the /api/v1/user_uploads API endpoint with ?api_key
        auth. Here a single file is uploaded and downloaded using a
        username and api_key
        """
        fp = StringIO("zulip!")
        fp.name = "zulip.txt"

        # Upload file via API
        result = self.api_post(self.example_user("hamlet"), "/api/v1/user_uploads", {"file": fp})
        response_dict = self.assert_json_success(result)
        self.assertIn("uri", response_dict)
        self.assertIn("url", response_dict)
        url = response_dict["url"]
        self.assertEqual(response_dict["uri"], url)
        base = "/user_uploads/"
        self.assertEqual(base, url[: len(base)])

        self.logout()

        # Try to download file via API, passing URL and invalid API key
        user_profile = self.example_user("hamlet")

        response = self.client_get(url, {"api_key": "invalid"})
        self.assertEqual(response.status_code, 401)

        response = self.client_get(url, {"api_key": user_profile.api_key})
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.getvalue(), b"zulip!")

    def test_file_too_big_failure(self) -> None:
        """
        Attempting to upload big files should fail.
        """
        self.login("hamlet")
        fp = StringIO("bah!")
        fp.name = "a.txt"

        # Use MAX_FILE_UPLOAD_SIZE of 0, because the next increment
        # would be 1MB.
        with self.settings(MAX_FILE_UPLOAD_SIZE=0):
            result = self.client_post("/json/user_uploads", {"f1": fp})
        self.assert_json_error(
            result,
            "File is larger than this server's configured maximum upload size (0 MiB).",
        )

    def test_file_too_big_failure_standard_plan(self) -> None:
        """
        Verify error message where a plan is involved.
        """
        self.login("hamlet")
        fp = StringIO("bah!")
        fp.name = "a.txt"

        realm = get_realm("zulip")
        realm.plan_type = Realm.PLAN_TYPE_LIMITED
        realm.save()
        with self.settings(MAX_FILE_UPLOAD_SIZE=0):
            result = self.client_post("/json/user_uploads", {"f1": fp})
        self.assert_json_error(
            result,
            "File is larger than the maximum upload size (0 MiB) allowed by your organization's plan.",
        )

    def test_multiple_upload_failure(self) -> None:
        """
        Attempting to upload two files should fail.
        """
        self.login("hamlet")
        fp = StringIO("bah!")
        fp.name = "a.txt"
        fp2 = StringIO("pshaw!")
        fp2.name = "b.txt"

        result = self.client_post("/json/user_uploads", {"f1": fp, "f2": fp2})
        self.assert_json_error(result, "You may only upload one file at a time")

    def test_no_file_upload_failure(self) -> None:
        """
        Calling this endpoint with no files should fail.
        """
        self.login("hamlet")

        result = self.client_post("/json/user_uploads")
        self.assert_json_error(result, "You must specify a file to upload")

    def test_guess_content_type_from_filename(self) -> None:
        """
        Test coverage for files without content-type in the metadata;
        in which case we try to guess the content-type from the filename.
        """
        uploaded_file = SimpleUploadedFile("somefile", b"zulip!", content_type="")
        result = self.api_post(
            self.example_user("hamlet"), "/api/v1/user_uploads", {"file": uploaded_file}
        )

        self.login("hamlet")
        response_dict = self.assert_json_success(result)
        url = response_dict["url"]
        result = self.client_get(url)
        self.assertEqual(result.status_code, 200)
        self.assertEqual(result["Content-Type"], "application/octet-stream")
        consume_response(result)

        # Old files may be stored without a content-type in the
        # database, in which case we try to guess at download time.
        attachment = Attachment.objects.get(file_name="somefile")
        self.assertEqual(attachment.content_type, "application/octet-stream")
        attachment.content_type = None
        attachment.save(update_fields=["content_type"])
        result = self.client_get(url)
        self.assertEqual(result.status_code, 200)
        self.assertEqual(result["Content-Type"], "application/octet-stream")
        consume_response(result)

        uploaded_file = SimpleUploadedFile("somefile.txt", b"zulip!", content_type="")
        result = self.api_post(
            self.example_user("hamlet"), "/api/v1/user_uploads", {"file": uploaded_file}
        )
        self.assert_json_success(result)

        response_dict = self.assert_json_success(result)
        url = response_dict["url"]
        result = self.client_get(url)
        self.assertEqual(result.status_code, 200)
        self.assertEqual(result["Content-Type"], 'text/plain; charset="ascii"')
        consume_response(result)

        # As above, test without a stored content_type
        attachment = Attachment.objects.get(file_name="somefile.txt")
        self.assertEqual(attachment.content_type, 'text/plain; charset="ascii"')
        attachment.content_type = None
        attachment.save(update_fields=["content_type"])
        result = self.client_get(url)
        self.assertEqual(result.status_code, 200)
        self.assertEqual(result["Content-Type"], 'text/plain; charset="ascii"')
        consume_response(result)

    def test_guess_content_type_charset(self) -> None:
        tests = [
            ("No high bytes in this string", "ascii", "ascii"),  # Explicit ASCII encoding
            ("नाम में क्या रक्खा हे", "utf-8", "utf-8"),  # Enough to get 99% confidence UTF-8
            ("日本語", "iso2022_jp", "ISO-2022-JP"),  # Non-UTF-8 99% confidence
            ("日本語", "utf-8", "utf-8"),  # UTF-8 is only 87% confident
            (
                "Min svävare är full av ålar.",
                "iso-8859-1",
                "ISO-8859-1",
            ),  # iso-8859-1 maxes out at 73%, but we'll still guess it
            ("Aucune idée", "mac-roman", None),  # Short text in obscure formats is left unguessed
        ]
        self.login("hamlet")
        for test_string, encoded_in, found_encoding in tests:
            with self.subTest(msg=f"Encoding '{test_string}' in {encoded_in}"):
                uploaded_file = SimpleUploadedFile(
                    "somefile.txt", test_string.encode(encoded_in), content_type="text/plain"
                )
                result = self.api_post(
                    self.example_user("hamlet"), "/api/v1/user_uploads", {"file": uploaded_file}
                )

                response_dict = self.assert_json_success(result)
                url = response_dict["url"]
                result = self.client_get(url)
                self.assertEqual(result.status_code, 200)
                expected = "text/plain"
                if found_encoding is not None:
                    expected += f'; charset="{found_encoding}"'
                self.assertEqual(result["Content-Type"], expected)
                consume_response(result)

    def test_content_type_charset_specified(self) -> None:
        # Setting the charset requires a NamedTemporaryFile, as
        # SimpleUploadedFile does not transmit a charset even if it's
        # provided as part of the content_type
        with tempfile.NamedTemporaryFile() as uploaded_file:
            uploaded_file.write("नाम में क्या रक्खा हे".encode())
            uploaded_file.seek(0)
            # We intentionally provide the _wrong_ charset on this, so
            # that we verify that the charset detection code is not
            # overriding the value that the user claims.
            uploaded_file.content_type = "text/plain; test-key=test_value; charset=big5"  # type: ignore[attr-defined]

            result = self.api_post(
                self.example_user("hamlet"), "/api/v1/user_uploads", {"file": uploaded_file}
            )

        self.login("hamlet")
        response_dict = self.assert_json_success(result)
        url = response_dict["url"]
        result = self.client_get(url)
        self.assertEqual(result.status_code, 200)
        self.assertEqual(
            result["Content-Type"], 'text/plain; test-key="test_value"; charset="big5"'
        )
        consume_response(result)

    # This test will go through the code path for uploading files onto LOCAL storage
    # when Zulip is in DEVELOPMENT mode.
    def test_file_upload_authed(self) -> None:
        """
        A call to /json/user_uploads should return a url and actually create an
        entry in the database. This entry will be marked unclaimed till a message
        refers it.
        """
        self.login("hamlet")
        fp = StringIO("zulip!")
        fp.name = "zulip.txt"

        result = self.client_post("/json/user_uploads", {"file": fp})
        response_dict = self.assert_json_success(result)
        self.assertIn("uri", response_dict)
        self.assertIn("url", response_dict)
        url = response_dict["url"]
        self.assertEqual(response_dict["uri"], url)
        base = "/user_uploads/"
        self.assertEqual(base, url[: len(base)])

        self.assertEqual(self.client_get(url).getvalue(), b"zulip!")

        # Check the download endpoint
        download_url = url.replace("/user_uploads/", "/user_uploads/download/")
        result = self.client_get(download_url)
        self.assertEqual(result.getvalue(), b"zulip!")
        self.assertIn("attachment;", result.headers["Content-Disposition"])

        # check if DB has attachment marked as unclaimed
        entry = Attachment.objects.get(file_name="zulip.txt")
        self.assertEqual(entry.is_claimed(), False)

        hamlet = self.example_user("hamlet")
        self.subscribe(hamlet, "Denmark")
        body = f"First message ...[zulip.txt]({hamlet.realm.host}{url})"
        self.send_stream_message(hamlet, "Denmark", body, "test")

        # Now try the endpoint that's supposed to return a temporary URL for access
        # to the file.
        result = self.client_get("/json" + url)
        data = self.assert_json_success(result)
        url_only_url = data["url"]
        # Ensure this is different from the original url:
        self.assertNotEqual(url_only_url, url)
        self.assertIn("user_uploads/temporary/", url_only_url)
        self.assertTrue(url_only_url.endswith("zulip.txt"))
        # The generated URL has a token authorizing the requester to access the file
        # without being logged in.
        self.logout()
        self.assertEqual(self.client_get(url_only_url).getvalue(), b"zulip!")
        # The original url shouldn't work when logged out -- it redirects to the login page
        result = self.client_get(url)
        self.assertEqual(result.status_code, 302)
        self.assertTrue(result.headers["Location"].endswith(f"/login/?next={url}"))

    def test_serve_file_unauthed(self) -> None:
        self.login("hamlet")
        fp = StringIO("zulip!")
        fp.name = "zulip_web_public.txt"

        result = self.client_post("/json/user_uploads", {"file": fp})
        url = self.assert_json_success(result)["url"]

        with ratelimit_rule(86400, 1000, domain="spectator_attachment_access_by_file"):
            # Deny file access for non-web-public stream
            self.subscribe(self.example_user("hamlet"), "Denmark")
            host = self.example_user("hamlet").realm.host
            body = f"First message ...[zulip.txt](http://{host}" + url + ")"
            self.send_stream_message(self.example_user("hamlet"), "Denmark", body, "test")

            self.logout()
            response = self.client_get(url)
            self.assertEqual(response.status_code, 302)
            self.assertTrue(response.headers["Location"].endswith(f"/login/?next={url}"))

            # Allow file access for web-public stream
            self.login("hamlet")
            self.make_stream("web-public-stream", is_web_public=True)
            self.subscribe(self.example_user("hamlet"), "web-public-stream")
            body = f"First message ...[zulip.txt](http://{host}" + url + ")"
            self.send_stream_message(self.example_user("hamlet"), "web-public-stream", body, "test")

            self.logout()
            response = self.client_get(url)
            self.assertEqual(response.status_code, 200)
            consume_response(response)

        # Deny file access since rate limited
        with ratelimit_rule(86400, 0, domain="spectator_attachment_access_by_file"):
            response = self.client_get(url)
            self.assertEqual(response.status_code, 302)
            self.assertTrue(response.headers["Location"].endswith(f"/login/?next={url}"))

        # Check that the /download/ variant works as well
        download_url = url.replace("/user_uploads/", "/user_uploads/download/")
        with ratelimit_rule(86400, 1000, domain="spectator_attachment_access_by_file"):
            response = self.client_get(download_url)
            self.assertEqual(response.status_code, 200)
            consume_response(response)
        with ratelimit_rule(86400, 0, domain="spectator_attachment_access_by_file"):
            response = self.client_get(download_url)
            self.assertEqual(response.status_code, 302)
            self.assertTrue(response.headers["Location"].endswith(f"/login/?next={download_url}"))

        # Deny random file access
        response = self.client_get(
            "/user_uploads/2/71/QYB7LA-ULMYEad-QfLMxmI2e/zulip-non-existent.txt"
        )
        self.assertEqual(response.status_code, 404)

    def test_serve_local_file_unauthed_invalid_token(self) -> None:
        result = self.client_get("/user_uploads/temporary/badtoken/file.png")
        self.assert_json_error(result, "Invalid token")

    def test_serve_local_file_unauthed_altered_filename(self) -> None:
        self.login("hamlet")
        fp = StringIO("zulip!")
        fp.name = "zulip.txt"
        result = self.client_post("/json/user_uploads", {"file": fp})
        response_dict = self.assert_json_success(result)
        url = "/json" + response_dict["url"]

        result = self.client_get(url)
        data = self.assert_json_success(result)
        url_only_url = data["url"]

        self.assertTrue(url_only_url.endswith("zulip.txt"))
        url_only_url_changed_filename = url_only_url.split("zulip.txt")[0] + "differentname.exe"
        result = self.client_get(url_only_url_changed_filename)
        self.assert_json_error(result, "Invalid filename")

    def test_serve_local_file_unauthed_token_expires(self) -> None:
        self.login("hamlet")
        fp = StringIO("zulip!")
        fp.name = "zulip.txt"
        now = timezone_now()
        with time_machine.travel(now, tick=False):
            result = self.client_post("/json/user_uploads", {"file": fp})
            response_dict = self.assert_json_success(result)
            url = "/json" + response_dict["url"]

            result = self.client_get(url)
            data = self.assert_json_success(result)
            url_only_url = data["url"]
            self.logout()
            self.assertEqual(self.client_get(url_only_url).getvalue(), b"zulip!")

        with time_machine.travel(now + timedelta(seconds=30), tick=False):
            self.assertEqual(self.client_get(url_only_url).getvalue(), b"zulip!")

        # After over 60 seconds, the token should become invalid:
        with time_machine.travel(now + timedelta(seconds=61), tick=False):
            result = self.client_get(url_only_url)
            self.assert_json_error(result, "Invalid token")

    def test_serve_local_file_unauthed_token_deleted(self) -> None:
        self.login("hamlet")
        fp = StringIO("zulip!")
        fp.name = "zulip.txt"
        now = timezone_now()
        with time_machine.travel(now, tick=False):
            result = self.client_post("/json/user_uploads", {"file": fp})
            response_dict = self.assert_json_success(result)
            url = "/json" + response_dict["url"]

            result = self.client_get(url)
            data = self.assert_json_success(result)
            url_only_url = data["url"]

            path_id = response_dict["url"].removeprefix("/user_uploads/")
            Attachment.objects.get(path_id=path_id).delete()

            result = self.client_get(url_only_url)
            self.assert_json_error(result, "Invalid token")

    def test_file_download_unauthed(self) -> None:
        self.login("hamlet")
        fp = StringIO("zulip!")
        fp.name = "zulip.txt"
        result = self.client_post("/json/user_uploads", {"file": fp})
        response_dict = self.assert_json_success(result)
        url = response_dict["url"]

        self.logout()
        response = self.client_get(url)
        self.assertEqual(response.status_code, 302)
        self.assertTrue(response.headers["Location"].endswith(f"/login/?next={url}"))

    def test_image_download_unauthed(self) -> None:
        """
        As the above, but with an Accept header that prefers images.
        """
        self.login("hamlet")
        fp = StringIO("zulip!")
        fp.name = "zulip.txt"
        result = self.client_post("/json/user_uploads", {"file": fp})
        response_dict = self.assert_json_success(result)
        url = response_dict["url"]

        self.logout()
        response = self.client_get(
            url,
            # This is what Chrome sends for <img> tags
            headers={"Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8"},
        )
        self.assertEqual(response.status_code, 403)
        self.assertEqual(response.headers["Content-Type"], "image/png")
        consume_response(response)

    def test_removed_file_download(self) -> None:
        """
        Trying to download deleted files should return 404 error
        """
        self.login("hamlet")
        fp = StringIO("zulip!")
        fp.name = "zulip.txt"
        result = self.client_post("/json/user_uploads", {"file": fp})
        response_dict = self.assert_json_success(result)

        assert settings.LOCAL_UPLOADS_DIR is not None
        self.rm_tree(settings.LOCAL_UPLOADS_DIR)

        response = self.client_get(response_dict["uri"])
        self.assertEqual(response.status_code, 404)
        response = self.client_get(response_dict["url"])
        self.assertEqual(response.status_code, 404)

    def test_non_existing_file_download(self) -> None:
        """
        Trying to download a file that was never uploaded will return a json_error
        """
        hamlet = self.example_user("hamlet")
        self.login_user(hamlet)
        response = self.client_get(
            f"http://{hamlet.realm.host}/user_uploads/{hamlet.realm_id}/ff/gg/abc.py"
        )
        self.assertEqual(response.status_code, 404)
        self.assert_in_response("This file does not exist or has been deleted.", response)

    def test_non_existing_image_download(self) -> None:
        """
        As the above method, but with an Accept header that prefers images to text
        """
        hamlet = self.example_user("hamlet")
        self.login_user(hamlet)
        response = self.client_get(
            f"http://{hamlet.realm.host}/user_uploads/{hamlet.realm_id}/ff/gg/abc.png",
            # This is what Chrome sends for <img> tags
            headers={"Accept": "image/avif,image/webp,image/apng,image/*,*/*;q=0.8"},
        )
        self.assertEqual(response.status_code, 404)
        self.assertEqual(response.headers["Content-Type"], "image/png")
        consume_response(response)

        response = self.client_get(
            f"http://{hamlet.realm.host}/user_uploads/{hamlet.realm_id}/ff/gg/abc.png",
            # Ask for something neither image nor text -- you get text as a default
            headers={"Accept": "audio/*,application/octet-stream"},
        )
        self.assertEqual(response.status_code, 404)
        self.assertEqual(response.headers["Content-Type"], "text/html; charset=utf-8")
        self.assert_in_response("This file does not exist or has been deleted.", response)

    def test_attachment_url_without_upload(self) -> None:
        hamlet = self.example_user("hamlet")
        self.login_user(hamlet)
        with self.assertLogs(level="WARNING") as warn_log:
            body = f"Test message ...[zulip.txt](http://{hamlet.realm.host}/user_uploads/{hamlet.realm_id}/64/fake_path_id.txt)"
            message_id = self.send_stream_message(
                self.example_user("hamlet"), "Denmark", body, "test"
            )
        self.assertFalse(
            Attachment.objects.filter(path_id=f"{hamlet.realm_id}/64/fake_path_id.txt").exists()
        )

        self.assertEqual(
            warn_log.output,
            [
                f"WARNING:root:User {hamlet.id} tried to share upload {hamlet.realm_id}/64/fake_path_id.txt in message {message_id}, but lacks permission"
            ],
        )

    def test_multiple_claim_attachments(self) -> None:
        """
        This test tries to claim the same attachment twice. The messages field in
        the Attachment model should have both the messages in its entry.
        """
        self.login("hamlet")
        d1 = StringIO("zulip!")
        d1.name = "dummy_1.txt"
        result = self.client_post("/json/user_uploads", {"file": d1})
        response_dict = self.assert_json_success(result)
        d1_path_id = re.sub(r"/user_uploads/", "", response_dict["url"])

        self.subscribe(self.example_user("hamlet"), "Denmark")
        host = self.example_user("hamlet").realm.host
        body = f"First message ...[zulip.txt](http://{host}/user_uploads/" + d1_path_id + ")"
        self.send_stream_message(self.example_user("hamlet"), "Denmark", body, "test")
        body = f"Second message ...[zulip.txt](http://{host}/user_uploads/" + d1_path_id + ")"
        self.send_stream_message(self.example_user("hamlet"), "Denmark", body, "test")

        self.assertEqual(Attachment.objects.get(path_id=d1_path_id).messages.count(), 2)

    def test_multiple_claim_attachments_different_owners(self) -> None:
        """This test tries to claim the same attachment more than once, first
        with a private stream and then with different recipients."""
        self.login("hamlet")
        d1 = StringIO("zulip!")
        d1.name = "dummy_1.txt"
        result = self.client_post("/json/user_uploads", {"file": d1})
        response_dict = self.assert_json_success(result)
        d1_path_id = re.sub(r"/user_uploads/", "", response_dict["url"])
        host = self.example_user("hamlet").realm.host

        self.make_stream("private_stream", invite_only=True)
        self.subscribe(self.example_user("hamlet"), "private_stream")

        # First, send the message to the new private stream.
        body = f"First message ...[zulip.txt](http://{host}/user_uploads/" + d1_path_id + ")"
        self.send_stream_message(self.example_user("hamlet"), "private_stream", body, "test")
        self.assertFalse(Attachment.objects.get(path_id=d1_path_id).is_realm_public)
        self.assertEqual(Attachment.objects.get(path_id=d1_path_id).messages.count(), 1)

        # Then, try having a user who didn't receive the message try to publish it, and fail
        body = f"Illegal message ...[zulip.txt](http://{host}/user_uploads/" + d1_path_id + ")"
        cordelia = self.example_user("cordelia")
        with self.assertLogs(level="WARNING") as warn_log:
            self.send_stream_message(cordelia, "Verona", body, "test")
        self.assertTrue(
            f"WARNING:root:User {cordelia.id} tried to share upload" in warn_log.output[0]
            and "but lacks permission" in warn_log.output[0]
        )
        self.assertEqual(Attachment.objects.get(path_id=d1_path_id).messages.count(), 1)
        self.assertFalse(Attachment.objects.get(path_id=d1_path_id).is_realm_public)
        self.assertFalse(Attachment.objects.get(path_id=d1_path_id).is_web_public)

        # Then, have it in a direct message to another user, giving that other user access.
        body = f"Second message ...[zulip.txt](http://{host}/user_uploads/" + d1_path_id + ")"
        self.send_personal_message(self.example_user("hamlet"), self.example_user("othello"), body)
        self.assertEqual(Attachment.objects.get(path_id=d1_path_id).messages.count(), 2)
        self.assertFalse(Attachment.objects.get(path_id=d1_path_id).is_realm_public)
        self.assertFalse(Attachment.objects.get(path_id=d1_path_id).is_web_public)

        # Then, have that new recipient user publish it.
        body = f"Third message ...[zulip.txt](http://{host}/user_uploads/" + d1_path_id + ")"
        self.send_stream_message(self.example_user("othello"), "Verona", body, "test")
        self.assertEqual(Attachment.objects.get(path_id=d1_path_id).messages.count(), 3)
        self.assertTrue(Attachment.objects.get(path_id=d1_path_id).is_realm_public)
        self.assertFalse(Attachment.objects.get(path_id=d1_path_id).is_web_public)

        # Finally send to Rome, the web-public stream, and confirm it's now web-public
        body = f"Fourth message ...[zulip.txt](http://{host}/user_uploads/" + d1_path_id + ")"
        self.subscribe(self.example_user("othello"), "Rome")
        self.send_stream_message(self.example_user("othello"), "Rome", body, "test")
        self.assertEqual(Attachment.objects.get(path_id=d1_path_id).messages.count(), 4)
        self.assertTrue(Attachment.objects.get(path_id=d1_path_id).is_realm_public)
        self.assertTrue(Attachment.objects.get(path_id=d1_path_id).is_web_public)

    def test_check_attachment_reference_update(self) -> None:
        f1 = StringIO("file1")
        f1.name = "file1.txt"
        f2 = StringIO("file2")
        f2.name = "file2.txt"
        f3 = StringIO("file3")
        f3.name = "file3.txt"
        hamlet = self.example_user("hamlet")
        host = hamlet.realm.host

        self.login_user(hamlet)
        result = self.client_post("/json/user_uploads", {"file": f1})
        response_dict = self.assert_json_success(result)
        f1_path_id = re.sub(r"/user_uploads/", "", response_dict["url"])

        result = self.client_post("/json/user_uploads", {"file": f2})
        response_dict = self.assert_json_success(result)
        f2_path_id = re.sub(r"/user_uploads/", "", response_dict["url"])

        self.subscribe(hamlet, "test")
        body = (
            f"[f1.txt](http://{host}/user_uploads/" + f1_path_id + ") "
            f"[f2.txt](http://{host}/user_uploads/" + f2_path_id + ")"
        )
        msg_id = self.send_stream_message(hamlet, "test", body, "test")

        result = self.client_post("/json/user_uploads", {"file": f3})
        response_dict = self.assert_json_success(result)
        f3_path_id = re.sub(r"/user_uploads/", "", response_dict["url"])

        new_body = (
            f"[f3.txt](http://{host}/user_uploads/" + f3_path_id + ") "
            f"[f2.txt](http://{host}/user_uploads/" + f2_path_id + ")"
        )
        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "content": new_body,
            },
        )
        self.assert_json_success(result)

        message = Message.objects.get(id=msg_id)
        f1_attachment = Attachment.objects.get(path_id=f1_path_id)
        f2_attachment = Attachment.objects.get(path_id=f2_path_id)
        f3_attachment = Attachment.objects.get(path_id=f3_path_id)

        self.assertTrue(message not in f1_attachment.messages.all())
        self.assertTrue(message in f2_attachment.messages.all())
        self.assertTrue(message in f3_attachment.messages.all())

        # Delete all the attachments from the message
        new_body = "(deleted)"
        result = self.client_patch(
            "/json/messages/" + str(msg_id),
            {
                "content": new_body,
            },
        )
        self.assert_json_success(result)

        message = Message.objects.get(id=msg_id)
        f1_attachment = Attachment.objects.get(path_id=f1_path_id)
        f2_attachment = Attachment.objects.get(path_id=f2_path_id)
        f3_attachment = Attachment.objects.get(path_id=f3_path_id)
        self.assertTrue(message not in f1_attachment.messages.all())
        self.assertTrue(message not in f2_attachment.messages.all())
        self.assertTrue(message not in f3_attachment.messages.all())

    def test_file_name(self) -> None:
        """
        Unicode filenames should be processed correctly.
        """
        self.login("hamlet")
        for expected in ["Здравейте.txt", "test"]:
            fp = StringIO("bah!")
            fp.name = quote(expected)

            result = self.client_post("/json/user_uploads", {"f1": fp})
            response_dict = self.assert_json_success(result)
            assert sanitize_name(expected) in response_dict["uri"]
            assert sanitize_name(expected) in response_dict["url"]

    def test_sanitize_file_name(self) -> None:
        self.login("hamlet")
        for uploaded_filename, expected_url, expected_filename in [
            ("../foo", "foo", "foo"),
            (".. ", "uploaded-file", ".. "),
            ("/", "f1", "f1"),
