stash
This commit is contained in:
parent
3b494dfa37
commit
0f51f6e33f
14 changed files with 534 additions and 356 deletions
|
|
@ -1,13 +1,20 @@
|
|||
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
|
||||
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()
|
||||
|
|
@ -105,6 +112,23 @@ class FilesTestCase(FilesTestMixin, ToolshedTestCase):
|
|||
self.assertEqual(File.objects.count(), 3)
|
||||
self.assertEqual(countdir(DefaultStorage(), ''), 3)
|
||||
|
||||
def test_file_upload_reclaims_stale_orphan_at_canonical_path(self):
|
||||
# Reproduces a real incident: a File row gets deleted without its underlying stored
|
||||
# bytes being removed (e.g. a bug in some cleanup call site), leaving an orphan sitting
|
||||
# at the exact path hash_upload() would compute for that content. A later upload of the
|
||||
# same content must land back on that canonical path - not get silently suffixed by
|
||||
# Django's default collision-avoidance, which would make it unreachable to every part
|
||||
# of the app that derives a file's URL purely from its hash.
|
||||
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):
|
||||
|
|
@ -120,28 +144,29 @@ class MediaUrlTestCase(FilesTestMixin, UserTestMixin, InventoryTestMixin, Toolsh
|
|||
self.f['item2'].files.add(self.f['test_file1'])
|
||||
|
||||
|
||||
# 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)
|
||||
@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'])
|
||||
|
|
@ -195,3 +220,138 @@ class MediaUrlTestCase(FilesTestMixin, UserTestMixin, InventoryTestMixin, Toolsh
|
|||
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 method gets its own distinct image content (and therefore its own content
|
||||
# hash / thumbnail cache path) - InMemoryStorage isn't reset between test methods within
|
||||
# a run, so sharing one fixed image across methods risks one test's cached (or, in
|
||||
# test_thumbnail_served_from_cache_on_second_request's case, deliberately corrupted)
|
||||
# thumbnail leaking into another test'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):
|
||||
# Documents the fixed size allow-list this test suite exercises against - update both
|
||||
# if files/media_urls.py's THUMBNAIL_SIZES ever 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):
|
||||
# Reproduces a real incident: an 'LA' (grayscale + alpha) source whose fully-transparent
|
||||
# region has zeroed-out luminance underneath, as many image tools produce. Converting
|
||||
# straight to RGB (dropping alpha without compositing) reveals that zeroed data - the
|
||||
# whole thumbnail comes out solid black even though the visible (opaque) content isn't.
|
||||
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()
|
||||
|
||||
# Overwrite the cached file with a marker so a correct implementation must serve this
|
||||
# exact content back rather than regenerating it from the original.
|
||||
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 in these fixtures (see prepare_inventory) - the
|
||||
# denied case needs a stranger to that friendship 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")
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue