mirror of
https://github.com/zulip/zulip.git
synced 2025-11-11 01:16:19 +00:00
typing: Add appropriate none-checks for LOCAL_UPLOADS_DIR.
This is a part of django-stubs refactorings. Signed-off-by: Zixuan James Li <359101898@qq.com>
This commit is contained in:
committed by
Tim Abbott
parent
d5517932cd
commit
bb6a934c8d
@@ -217,6 +217,7 @@ def avatar_disk_path(
|
|||||||
) -> str:
|
) -> str:
|
||||||
avatar_url_path = avatar_url(user_profile, medium)
|
avatar_url_path = avatar_url(user_profile, medium)
|
||||||
assert avatar_url_path is not None
|
assert avatar_url_path is not None
|
||||||
|
assert settings.LOCAL_UPLOADS_DIR is not None
|
||||||
avatar_disk_path = os.path.join(
|
avatar_disk_path = os.path.join(
|
||||||
settings.LOCAL_UPLOADS_DIR,
|
settings.LOCAL_UPLOADS_DIR,
|
||||||
"avatars",
|
"avatars",
|
||||||
|
|||||||
@@ -236,6 +236,7 @@ def init_worker(counter: "multiprocessing.sharedctypes.Synchronized[int]") -> No
|
|||||||
# We manually update the upload directory path in the URL regex.
|
# We manually update the upload directory path in the URL regex.
|
||||||
from zproject.dev_urls import avatars_url
|
from zproject.dev_urls import avatars_url
|
||||||
|
|
||||||
|
assert settings.LOCAL_UPLOADS_DIR is not None
|
||||||
new_root = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars")
|
new_root = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars")
|
||||||
avatars_url.default_args["document_root"] = new_root
|
avatars_url.default_args["document_root"] = new_root
|
||||||
|
|
||||||
|
|||||||
@@ -23,6 +23,7 @@ def transfer_uploads_to_s3(processes: int) -> None:
|
|||||||
|
|
||||||
def _transfer_avatar_to_s3(user: UserProfile) -> None:
|
def _transfer_avatar_to_s3(user: UserProfile) -> None:
|
||||||
avatar_path = user_avatar_path(user)
|
avatar_path = user_avatar_path(user)
|
||||||
|
assert settings.LOCAL_UPLOADS_DIR is not None
|
||||||
file_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", avatar_path) + ".original"
|
file_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", avatar_path) + ".original"
|
||||||
try:
|
try:
|
||||||
with open(file_path, "rb") as f:
|
with open(file_path, "rb") as f:
|
||||||
@@ -46,6 +47,7 @@ def transfer_avatars_to_s3(processes: int) -> None:
|
|||||||
|
|
||||||
|
|
||||||
def _transfer_message_files_to_s3(attachment: Attachment) -> None:
|
def _transfer_message_files_to_s3(attachment: Attachment) -> None:
|
||||||
|
assert settings.LOCAL_UPLOADS_DIR is not None
|
||||||
file_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "files", attachment.path_id)
|
file_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "files", attachment.path_id)
|
||||||
try:
|
try:
|
||||||
with open(file_path, "rb") as f:
|
with open(file_path, "rb") as f:
|
||||||
@@ -82,6 +84,7 @@ def _transfer_emoji_to_s3(realm_emoji: RealmEmoji) -> None:
|
|||||||
realm_id=realm_emoji.realm.id,
|
realm_id=realm_emoji.realm.id,
|
||||||
emoji_file_name=realm_emoji.file_name,
|
emoji_file_name=realm_emoji.file_name,
|
||||||
)
|
)
|
||||||
|
assert settings.LOCAL_UPLOADS_DIR is not None
|
||||||
emoji_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", emoji_path) + ".original"
|
emoji_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", emoji_path) + ".original"
|
||||||
try:
|
try:
|
||||||
with open(emoji_path, "rb") as f:
|
with open(emoji_path, "rb") as f:
|
||||||
|
|||||||
@@ -50,6 +50,7 @@ class LocalUploader(Uploader):
|
|||||||
os.makedirs(dirname)
|
os.makedirs(dirname)
|
||||||
|
|
||||||
def copy_files(self, src_path: str, dst_path: str) -> None:
|
def copy_files(self, src_path: str, dst_path: str) -> None:
|
||||||
|
assert settings.LOCAL_UPLOADS_DIR is not None
|
||||||
src_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", src_path)
|
src_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", src_path)
|
||||||
self.mkdirs(src_path)
|
self.mkdirs(src_path)
|
||||||
dst_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", dst_path)
|
dst_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", dst_path)
|
||||||
|
|||||||
@@ -1177,6 +1177,7 @@ class RealmImportExportTest(ExportFile):
|
|||||||
uploaded_file = Attachment.objects.get(realm=imported_realm)
|
uploaded_file = Attachment.objects.get(realm=imported_realm)
|
||||||
self.assert_length(b"zulip!", uploaded_file.size)
|
self.assert_length(b"zulip!", uploaded_file.size)
|
||||||
|
|
||||||
|
assert settings.LOCAL_UPLOADS_DIR is not None
|
||||||
attachment_file_path = os.path.join(
|
attachment_file_path = os.path.join(
|
||||||
settings.LOCAL_UPLOADS_DIR, "files", uploaded_file.path_id
|
settings.LOCAL_UPLOADS_DIR, "files", uploaded_file.path_id
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -1762,6 +1762,7 @@ class LocalStorageTest(UploadSerializeMixin, ZulipTestCase):
|
|||||||
base = "/user_uploads/"
|
base = "/user_uploads/"
|
||||||
self.assertEqual(base, uri[: len(base)])
|
self.assertEqual(base, uri[: len(base)])
|
||||||
path_id = re.sub("/user_uploads/", "", uri)
|
path_id = re.sub("/user_uploads/", "", uri)
|
||||||
|
assert settings.LOCAL_UPLOADS_DIR is not None
|
||||||
file_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "files", path_id)
|
file_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "files", path_id)
|
||||||
self.assertTrue(os.path.isfile(file_path))
|
self.assertTrue(os.path.isfile(file_path))
|
||||||
|
|
||||||
@@ -1800,6 +1801,7 @@ class LocalStorageTest(UploadSerializeMixin, ZulipTestCase):
|
|||||||
|
|
||||||
write_local_file("avatars", file_path + ".original", read_test_image_file("img.png"))
|
write_local_file("avatars", file_path + ".original", read_test_image_file("img.png"))
|
||||||
|
|
||||||
|
assert settings.LOCAL_UPLOADS_DIR is not None
|
||||||
image_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", file_path + ".original")
|
image_path = os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", file_path + ".original")
|
||||||
with open(image_path, "rb") as f:
|
with open(image_path, "rb") as f:
|
||||||
image_data = f.read()
|
image_data = f.read()
|
||||||
@@ -1878,10 +1880,12 @@ class LocalStorageTest(UploadSerializeMixin, ZulipTestCase):
|
|||||||
user_profile = self.example_user("iago")
|
user_profile = self.example_user("iago")
|
||||||
self.assertTrue(user_profile.is_realm_admin)
|
self.assertTrue(user_profile.is_realm_admin)
|
||||||
|
|
||||||
|
assert settings.TEST_WORKER_DIR is not None
|
||||||
tarball_path = os.path.join(settings.TEST_WORKER_DIR, "tarball.tar.gz")
|
tarball_path = os.path.join(settings.TEST_WORKER_DIR, "tarball.tar.gz")
|
||||||
with open(tarball_path, "w") as f:
|
with open(tarball_path, "w") as f:
|
||||||
f.write("dummy")
|
f.write("dummy")
|
||||||
|
|
||||||
|
assert settings.LOCAL_UPLOADS_DIR is not None
|
||||||
uri = upload_export_tarball(user_profile.realm, tarball_path)
|
uri = upload_export_tarball(user_profile.realm, tarball_path)
|
||||||
self.assertTrue(
|
self.assertTrue(
|
||||||
os.path.isfile(os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", tarball_path))
|
os.path.isfile(os.path.join(settings.LOCAL_UPLOADS_DIR, "avatars", tarball_path))
|
||||||
|
|||||||
Reference in New Issue
Block a user