49 lines
2.0 KiB
Python
49 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pathlib import Path
|
|
|
|
from web.routers.files import _numbered_upload_name, _replace_upload, _write_upload_without_overwrite
|
|
|
|
|
|
class FileUploadConflictTests(unittest.TestCase):
|
|
def test_number_is_inserted_before_complete_extension(self):
|
|
self.assertEqual(_numbered_upload_name("data.tar.gz", 2), "data (2).tar.gz")
|
|
self.assertEqual(_numbered_upload_name("报告", 1), "报告 (1)")
|
|
self.assertEqual(_numbered_upload_name(".env", 1), ".env (1)")
|
|
|
|
def test_existing_file_is_preserved(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
(root / "报告.pdf").write_bytes(b"old")
|
|
saved = _write_upload_without_overwrite(root, "报告.pdf", b"new")
|
|
self.assertEqual(saved.name, "报告 (1).pdf")
|
|
self.assertEqual((root / "报告.pdf").read_bytes(), b"old")
|
|
self.assertEqual(saved.read_bytes(), b"new")
|
|
|
|
def test_concurrent_uploads_get_distinct_names(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
saved = list(pool.map(
|
|
lambda i: _write_upload_without_overwrite(root, "same.txt", str(i).encode()),
|
|
range(4),
|
|
))
|
|
self.assertEqual(len({path.name for path in saved}), 4)
|
|
self.assertEqual(len(list(root.glob("same*.txt"))), 4)
|
|
|
|
def test_explicit_replace_changes_only_target(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
target = root / "target.txt"
|
|
target.write_bytes(b"old")
|
|
_replace_upload(target, b"new")
|
|
self.assertEqual(target.read_bytes(), b"new")
|
|
self.assertFalse(list(root.glob(".zcbot-upload-*")))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|