343 lines
16 KiB
Python
343 lines
16 KiB
Python
import io
|
|
import os
|
|
import zlib
|
|
|
|
from django.conf import settings
|
|
from django.core.files.base import ContentFile
|
|
from django.core.files.storage import DefaultStorage, default_storage
|
|
from django.db import IntegrityError, transaction
|
|
from django.test import Client, override_settings
|
|
from authentication.tests import SignatureAuthClient, ToolshedTestCase, UserTestMixin
|
|
from toolshed.tests import InventoryTestMixin
|
|
from nacl.hash import sha256
|
|
from nacl.encoding import HexEncoder
|
|
from PIL import Image
|
|
import base64
|
|
|
|
from files.media_urls import THUMBNAIL_SIZES
|
|
from files.models import File
|
|
|
|
anonymous_client = Client()
|
|
client = SignatureAuthClient()
|
|
|
|
|
|
def rmdir(storage, path):
|
|
dirs, files = storage.listdir(path)
|
|
for file in files:
|
|
storage.delete(path + file)
|
|
for dir in dirs:
|
|
rmdir(storage, path + dir + "/")
|
|
storage.delete(path)
|
|
|
|
|
|
def countdir(storage, path):
|
|
dirs, files = storage.listdir(path)
|
|
count = len(files)
|
|
for dir in dirs:
|
|
count += countdir(storage, path + dir + "/")
|
|
return count
|
|
|
|
|
|
class FilesTestMixin:
|
|
def prepare_files(self):
|
|
rmdir(DefaultStorage(), '')
|
|
self.f['test_content1'] = b'testcontent1'
|
|
self.f['hash1'] = sha256(self.f['test_content1'], encoder=HexEncoder).decode('utf-8')
|
|
self.f['encoded_content1'] = base64.b64encode(self.f['test_content1']).decode('utf-8')
|
|
self.f['test_file1'] = File.objects.create(mime_type='text/plain', data=self.f['encoded_content1'])
|
|
self.f['test_content2'] = b'testcontent2'
|
|
self.f['hash2'] = sha256(self.f['test_content2'], encoder=HexEncoder).decode('utf-8')
|
|
self.f['encoded_content2'] = base64.b64encode(self.f['test_content2']).decode('utf-8')
|
|
self.f['test_file2'] = File.objects.create(mime_type='text/plain', data=self.f['encoded_content2'])
|
|
self.f['test_content3'] = b'testcontent3'
|
|
self.f['hash3'] = sha256(self.f['test_content3'], encoder=HexEncoder).decode('utf-8')
|
|
self.f['encoded_content3'] = base64.b64encode(self.f['test_content3']).decode('utf-8')
|
|
self.f['test_file3'] = File.objects.create(mime_type='text/plain', data=self.f['encoded_content3'])
|
|
self.f['test_content4'] = b'testcontent4'
|
|
self.f['hash4'] = sha256(self.f['test_content4'], encoder=HexEncoder).decode('utf-8')
|
|
self.f['encoded_content4'] = base64.b64encode(self.f['test_content4']).decode('utf-8')
|
|
|
|
|
|
class FilesTestCase(FilesTestMixin, ToolshedTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.prepare_files()
|
|
|
|
def test_file_list(self):
|
|
self.assertEqual(File.objects.count(), 3)
|
|
self.assertEqual(countdir(DefaultStorage(), ''), 3)
|
|
|
|
def test_file_upload(self):
|
|
File.objects.create(mime_type='text/plain', data=self.f['encoded_content4'])
|
|
self.assertEqual(File.objects.count(), 4)
|
|
self.assertEqual(countdir(DefaultStorage(), ''), 4)
|
|
self.assertEqual(File.objects.get(id=4).file.read(), self.f['test_content4'])
|
|
self.assertEqual(File.objects.get(id=4).file.name,
|
|
f"{self.f['hash4'][:2]}/{self.f['hash4'][2:4]}/{self.f['hash4'][4:6]}/{self.f['hash4'][6:]}")
|
|
|
|
def test_file_upload_fail(self):
|
|
with transaction.atomic():
|
|
with self.assertRaises(ValueError):
|
|
File.objects.create(file=ContentFile(self.f['test_content4']), mime_type='text/plain')
|
|
self.assertEqual(File.objects.count(), 3)
|
|
self.assertEqual(countdir(DefaultStorage(), ''), 3)
|
|
|
|
def test_file_upload_duplicate(self):
|
|
with transaction.atomic():
|
|
with self.assertRaises(IntegrityError):
|
|
File.objects.create(mime_type='text/plain', data=self.f['encoded_content3'])
|
|
self.assertEqual(File.objects.count(), 3)
|
|
self.assertEqual(countdir(DefaultStorage(), ''), 3)
|
|
|
|
def test_file_upload_get_or_create(self):
|
|
file, created = File.objects.get_or_create(data=self.f['encoded_content3'])
|
|
self.assertEqual(File.objects.count(), 3)
|
|
self.assertEqual(countdir(DefaultStorage(), ''), 3)
|
|
self.assertFalse(created)
|
|
self.assertEqual(file.file.read(), self.f['test_content3'])
|
|
self.assertEqual(file.file.name,
|
|
f"{self.f['hash3'][:2]}/{self.f['hash3'][2:4]}/{self.f['hash3'][4:6]}/{self.f['hash3'][6:]}")
|
|
file, created = File.objects.get_or_create(data=self.f['encoded_content4'])
|
|
self.assertEqual(File.objects.count(), 4)
|
|
self.assertEqual(countdir(DefaultStorage(), ''), 4)
|
|
self.assertTrue(created)
|
|
self.assertEqual(file.file.read(), self.f['test_content4'])
|
|
self.assertEqual(file.file.name,
|
|
f"{self.f['hash4'][:2]}/{self.f['hash4'][2:4]}/{self.f['hash4'][4:6]}/{self.f['hash4'][6:]}")
|
|
|
|
def test_file_upload_get_or_create_fail(self):
|
|
with transaction.atomic():
|
|
with self.assertRaises(ValueError):
|
|
File.objects.get_or_create(hash=self.f['hash3'])
|
|
self.assertEqual(File.objects.count(), 3)
|
|
self.assertEqual(countdir(DefaultStorage(), ''), 3)
|
|
|
|
def test_file_upload_reclaims_stale_orphan_at_canonical_path(self):
|
|
# Regression test for a stale orphan at the canonical hash path. See docs/implementation.md#stale-orphan-cleanup-at-the-canonical-hash-path.
|
|
expected_path = f"{self.f['hash4'][:2]}/{self.f['hash4'][2:4]}/{self.f['hash4'][4:6]}/{self.f['hash4'][6:]}"
|
|
default_storage.save(expected_path, ContentFile(self.f['test_content4']))
|
|
self.assertTrue(default_storage.exists(expected_path))
|
|
self.assertFalse(File.objects.filter(hash=self.f['hash4']).exists())
|
|
|
|
file = File.objects.create(mime_type='text/plain', data=self.f['encoded_content4'])
|
|
|
|
self.assertEqual(file.file.name, expected_path)
|
|
self.assertEqual(file.file.read(), self.f['test_content4'])
|
|
|
|
|
|
class MediaUrlTestCase(FilesTestMixin, UserTestMixin, InventoryTestMixin, ToolshedTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.prepare_files()
|
|
self.prepare_users()
|
|
self.prepare_categories()
|
|
self.prepare_tags()
|
|
self.prepare_properties()
|
|
self.prepare_inventory()
|
|
self.f['item1'].files.add(self.f['test_file1'])
|
|
self.f['item1'].files.add(self.f['test_file2'])
|
|
self.f['item2'].files.add(self.f['test_file1'])
|
|
|
|
|
|
@override_settings(SERVE_X_ACCEL_REDIRECT=True)
|
|
def test_file_url(self):
|
|
reply = client.get(
|
|
f"/media/{self.f['hash1'][:2]}/{self.f['hash1'][2:4]}/{self.f['hash1'][4:6]}/{self.f['hash1'][6:]}",
|
|
self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
self.assertEqual(reply.headers['X-Accel-Redirect'],
|
|
f"/redirect_media/{self.f['hash1'][:2]}/{self.f['hash1'][2:4]}/{self.f['hash1'][4:6]}/{self.f['hash1'][6:]}")
|
|
self.assertEqual(reply.headers['Content-Type'], self.f['test_file1'].mime_type)
|
|
reply = client.get(
|
|
f"/media/{self.f['hash2'][:2]}/{self.f['hash2'][2:4]}/{self.f['hash2'][4:6]}/{self.f['hash2'][6:]}",
|
|
self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
self.assertEqual(reply.headers['X-Accel-Redirect'],
|
|
f"/redirect_media/{self.f['hash2'][:2]}/{self.f['hash2'][2:4]}/{self.f['hash2'][4:6]}/{self.f['hash2'][6:]}")
|
|
self.assertEqual(reply.headers['Content-Type'], self.f['test_file2'].mime_type)
|
|
reply = client.get(
|
|
f"/media/{self.f['hash2'][:2]}/{self.f['hash2'][2:4]}/{self.f['hash2'][4:6]}/{self.f['hash2'][6:]}",
|
|
self.f['local_user2'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
self.assertEqual(reply.headers['X-Accel-Redirect'],
|
|
f"/redirect_media/{self.f['hash2'][:2]}/{self.f['hash2'][2:4]}/{self.f['hash2'][4:6]}/{self.f['hash2'][6:]}")
|
|
self.assertEqual(reply.headers['Content-Type'], self.f['test_file2'].mime_type)
|
|
|
|
def test_file_url_fail(self):
|
|
reply = client.get('/media/{}/'.format('nonexistent'), self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 404)
|
|
self.assertTrue('X-Accel-Redirect' not in reply.headers)
|
|
|
|
def test_file_url_anonymous(self):
|
|
reply = anonymous_client.get(
|
|
f"/media/{self.f['hash1'][:2]}/{self.f['hash1'][2:4]}/{self.f['hash1'][4:6]}/{self.f['hash1'][6:]}")
|
|
self.assertEqual(reply.status_code, 403)
|
|
self.assertTrue('X-Accel-Redirect' not in reply.headers)
|
|
|
|
def test_file_url_wrong_user(self):
|
|
reply = client.get(
|
|
f"/media/{self.f['hash3'][:2]}/{self.f['hash3'][2:4]}/{self.f['hash3'][4:6]}/{self.f['hash3'][6:]}",
|
|
self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 404)
|
|
self.assertTrue('X-Accel-Redirect' not in reply.headers)
|
|
reply = client.get(
|
|
f"/media/{self.f['hash2'][:2]}/{self.f['hash2'][2:4]}/{self.f['hash2'][4:6]}/{self.f['hash2'][6:]}",
|
|
self.f['ext_user1'])
|
|
self.assertEqual(reply.status_code, 404)
|
|
self.assertTrue('X-Accel-Redirect' not in reply.headers)
|
|
|
|
@override_settings(SERVE_X_ACCEL_REDIRECT=True)
|
|
def test_profile_picture_url(self):
|
|
self.f['local_user1'].profile_picture = self.f['test_file3']
|
|
self.f['local_user1'].save()
|
|
|
|
reply = client.get(
|
|
f"/media/{self.f['hash3'][:2]}/{self.f['hash3'][2:4]}/{self.f['hash3'][4:6]}/{self.f['hash3'][6:]}",
|
|
self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
|
|
@override_settings(SERVE_X_ACCEL_REDIRECT=True)
|
|
def test_profile_picture_url_friend(self):
|
|
self.f['local_user1'].profile_picture = self.f['test_file3']
|
|
self.f['local_user1'].save()
|
|
|
|
reply = client.get(
|
|
f"/media/{self.f['hash3'][:2]}/{self.f['hash3'][2:4]}/{self.f['hash3'][4:6]}/{self.f['hash3'][6:]}",
|
|
self.f['local_user2'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
|
|
def test_profile_picture_url_not_friend(self):
|
|
self.f['local_user1'].profile_picture = self.f['test_file3']
|
|
self.f['local_user1'].save()
|
|
|
|
reply = client.get(
|
|
f"/media/{self.f['hash3'][:2]}/{self.f['hash3'][2:4]}/{self.f['hash3'][4:6]}/{self.f['hash3'][6:]}",
|
|
self.f['ext_user1'])
|
|
self.assertEqual(reply.status_code, 404)
|
|
|
|
|
|
class ThumbnailUrlTestCase(FilesTestMixin, UserTestMixin, InventoryTestMixin, ToolshedTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.prepare_files()
|
|
self.prepare_users()
|
|
self.prepare_categories()
|
|
self.prepare_tags()
|
|
self.prepare_properties()
|
|
self.prepare_inventory()
|
|
|
|
# Each test uses a distinct seeded image (own hash/cache path) since InMemoryStorage
|
|
# isn't reset between test methods, so a shared image risks one test's cached thumbnail leaking into another's assertions.
|
|
seed = zlib.crc32(self._testMethodName.encode()) % 256
|
|
buffer = io.BytesIO()
|
|
Image.new('RGB', (800, 600), (seed, 255 - seed, 128)).save(buffer, 'PNG')
|
|
image_bytes = buffer.getvalue()
|
|
self.f['image_hash'] = sha256(image_bytes, encoder=HexEncoder).decode('utf-8')
|
|
self.f['image_file'] = File.objects.create(
|
|
mime_type='image/png', data=base64.b64encode(image_bytes).decode('utf-8'))
|
|
self.f['item1'].files.add(self.f['image_file'])
|
|
|
|
def _thumb_url(self, size, image_hash=None):
|
|
h = image_hash or self.f['image_hash']
|
|
return f"/media/{size}/{h[:2]}/{h[2:4]}/{h[4:6]}/{h[6:]}/"
|
|
|
|
def _thumb_rel_path(self, size):
|
|
h = self.f['image_hash']
|
|
return os.path.join('thumbnails', str(size), h[:2], h[2:4], h[4:6], h[6:] + '.jpg')
|
|
|
|
def test_thumbnail_sizes_available(self):
|
|
# Fixed size allow-list this suite exercises - update both if media_urls.py's THUMBNAIL_SIZES changes.
|
|
self.assertEqual(THUMBNAIL_SIZES, (32, 64, 256))
|
|
|
|
@override_settings(SERVE_X_ACCEL_REDIRECT=False)
|
|
def test_thumbnail_generates_resized_jpeg(self):
|
|
self.assertFalse(default_storage.exists(self._thumb_rel_path(64)))
|
|
|
|
reply = client.get(self._thumb_url(64), self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
self.assertEqual(reply.headers['Content-Type'], 'image/jpeg')
|
|
|
|
generated = Image.open(io.BytesIO(reply.content))
|
|
self.assertEqual(generated.format, 'JPEG')
|
|
# Aspect-ratio-preserving fit within a 64x64 box, not a crop to exactly 64x64.
|
|
self.assertLessEqual(max(generated.size), 64)
|
|
self.assertAlmostEqual(generated.size[0] / generated.size[1], 800 / 600, places=2)
|
|
|
|
@override_settings(SERVE_X_ACCEL_REDIRECT=False)
|
|
def test_thumbnail_flattens_transparency_instead_of_going_black(self):
|
|
# Regression test for an 'LA' source with zeroed transparent-region luminance. See docs/implementation.md#rgba-flattening-avoids-revealing-black-under-transparent-pixels.
|
|
half_transparent = Image.new('LA', (200, 200))
|
|
pixels = half_transparent.load()
|
|
for x in range(200):
|
|
for y in range(200):
|
|
if x < 100:
|
|
pixels[x, y] = (0, 0) # transparent, zeroed-out luminance underneath
|
|
else:
|
|
pixels[x, y] = (255, 255) # fully opaque, bright content
|
|
|
|
buffer = io.BytesIO()
|
|
half_transparent.save(buffer, 'PNG')
|
|
image_bytes = buffer.getvalue()
|
|
image_hash = sha256(image_bytes, encoder=HexEncoder).decode('utf-8')
|
|
image_file = File.objects.create(
|
|
mime_type='image/png', data=base64.b64encode(image_bytes).decode('utf-8'))
|
|
self.f['item1'].files.add(image_file)
|
|
|
|
reply = client.get(self._thumb_url(64, image_hash=image_hash), self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
|
|
generated = Image.open(io.BytesIO(reply.content)).convert('L')
|
|
# The opaque (right) half must stay bright; a naive RGB conversion would blacken it too.
|
|
self.assertGreater(generated.getpixel((generated.width - 1, generated.height // 2)), 200)
|
|
self.assertNotEqual(generated.getextrema(), (0, 0))
|
|
|
|
@override_settings(SERVE_X_ACCEL_REDIRECT=False)
|
|
def test_thumbnail_served_from_cache_on_second_request(self):
|
|
client.get(self._thumb_url(64), self.f['local_user1'])
|
|
rel_path = self._thumb_rel_path(64)
|
|
with default_storage.open(rel_path, 'rb') as f:
|
|
cached_bytes = f.read()
|
|
|
|
# Overwrites the cache with a marker so a correct implementation must serve it back, not regenerate.
|
|
default_storage.delete(rel_path)
|
|
default_storage.save(rel_path, ContentFile(cached_bytes + b'MARKER'))
|
|
|
|
reply = client.get(self._thumb_url(64), self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
self.assertTrue(reply.content.endswith(b'MARKER'))
|
|
|
|
def test_thumbnail_invalid_size(self):
|
|
reply = client.get(self._thumb_url(100), self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 404)
|
|
self.assertFalse(default_storage.exists(self._thumb_rel_path(100)))
|
|
|
|
def test_thumbnail_not_found(self):
|
|
reply = client.get(self._thumb_url(64, image_hash='0' * 64), self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 404)
|
|
|
|
def test_thumbnail_anonymous(self):
|
|
reply = anonymous_client.get(self._thumb_url(64))
|
|
self.assertEqual(reply.status_code, 403)
|
|
|
|
def test_thumbnail_not_friend(self):
|
|
# local_user1/local_user2 are friends here (see prepare_inventory), so the denied case needs a stranger instead.
|
|
reply = client.get(self._thumb_url(64), self.f['ext_user1'])
|
|
self.assertEqual(reply.status_code, 404)
|
|
self.assertFalse(default_storage.exists(self._thumb_rel_path(64)))
|
|
|
|
def test_thumbnail_conditional_get(self):
|
|
reply = client.get(self._thumb_url(64), self.f['local_user1'])
|
|
etag = reply.headers['ETag']
|
|
self.assertEqual(etag, f"{self.f['image_hash']}_64")
|
|
|
|
reply = client.get(self._thumb_url(64), self.f['local_user1'], HTTP_IF_NONE_MATCH=etag)
|
|
self.assertEqual(reply.status_code, 304)
|
|
|
|
@override_settings(SERVE_X_ACCEL_REDIRECT=True)
|
|
def test_thumbnail_x_accel_redirect(self):
|
|
reply = client.get(self._thumb_url(64), self.f['local_user1'])
|
|
self.assertEqual(reply.status_code, 200)
|
|
h = self.f['image_hash']
|
|
self.assertEqual(reply.headers['X-Accel-Redirect'],
|
|
f"/redirect_media/thumbnails/64/{h[:2]}/{h[2:4]}/{h[4:6]}/{h[6:]}.jpg")
|
|
|