Closes #22169: Cache image file size on ImageAttachment (#22465)

This commit is contained in:
Jason Novinger 2026-06-23 10:40:19 -05:00 committed by GitHub
parent bf78a45204
commit d7c566aee6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 560 additions and 14 deletions

View File

@ -0,0 +1,102 @@
# Management Commands
In addition to Django's built-in management commands, NetBox provides several commands of its own. These are run using `manage.py`:
```
cd /opt/netbox
source /opt/netbox/venv/bin/activate
python3 netbox/manage.py <command>
```
Run any command with `--help` to see its full set of arguments.
## calculate_cached_counts
Force a recalculation of all cached counter fields (for example, the device count shown on a site). NetBox keeps these counters current automatically; this command is useful to repair them if they have drifted.
```
python3 netbox/manage.py calculate_cached_counts
```
## nbshell
Start the Django shell with all NetBox models already imported. See [NetBox Shell](./netbox-shell.md) for details.
```
python3 netbox/manage.py nbshell
```
## populate_image_sizes
!!! info "This command was introduced in NetBox v4.6.4."
Populate the cached file size for image attachments that predate the `image_size` field. Running this once after upgrading is recommended for deployments with many existing attachments on a remote storage backend (such as S3). It is safe to run on a live system and may be re-run; any file that cannot be read is skipped and retried on the next run.
```
python3 netbox/manage.py populate_image_sizes
```
## rebuild_prefixes
Rebuild the IPAM prefix hierarchy, recalculating the depth and child counts for all prefixes.
```
python3 netbox/manage.py rebuild_prefixes
```
## reindex
Reindex objects for the search backend. Pass one or more apps or models to reindex a subset; with no arguments, all models are reindexed. See [Removing a Plugin](../plugins/removal.md) for a related use.
```
python3 netbox/manage.py reindex [app_label[.ModelName] ...]
```
## renaturalize
Recalculate natural ordering values for the affected models. Pass one or more `app_label.ModelName` arguments to limit the scope; with no arguments, all models with natural ordering fields are processed.
```
python3 netbox/manage.py renaturalize [app_label.ModelName ...]
```
## runscript
Run a [custom script](../customization/custom-scripts.md) from the command line, outside the web UI or API.
```
python3 netbox/manage.py runscript <module.ScriptName>
```
## rqworker
Start a background task worker to process queued jobs (provided by django-rq). At least one worker must be running for background tasks such as report and script execution, webhooks, and synchronization to be processed.
```
python3 netbox/manage.py rqworker
```
## syncdatasource
Synchronize a data source from its remote upstream. Pass one or more data source names, or `--all` to synchronize every data source.
```
python3 netbox/manage.py syncdatasource <name> [<name> ...]
python3 netbox/manage.py syncdatasource --all
```
## trace_paths
Generate any missing cable paths among all cable termination objects. This is useful after a bulk import of cabling, or to repair paths that were not generated automatically.
```
python3 netbox/manage.py trace_paths
```
## webhook_receiver
Start a simple HTTP listener that prints any requests it receives. This is a debugging aid for testing webhooks: point a webhook at the listener and inspect exactly what NetBox sends. It listens on port 9000 by default; pass `--port` to change it and `--no-headers` to suppress the request headers.
```
python3 netbox/manage.py webhook_receiver [--port PORT] [--no-headers]
```

View File

@ -170,6 +170,7 @@ nav:
- Okta: 'administration/authentication/okta.md' - Okta: 'administration/authentication/okta.md'
- Permissions: 'administration/permissions.md' - Permissions: 'administration/permissions.md'
- Error Reporting: 'administration/error-reporting.md' - Error Reporting: 'administration/error-reporting.md'
- Management Commands: 'administration/management-commands.md'
- Replicating NetBox: 'administration/replicating-netbox.md' - Replicating NetBox: 'administration/replicating-netbox.md'
- NetBox Shell: 'administration/netbox-shell.md' - NetBox Shell: 'administration/netbox-shell.md'
- Data Model: - Data Model:

View File

@ -19,12 +19,13 @@ class ImageAttachmentSerializer(ValidatedModelSerializer):
parent = GFKSerializerField(read_only=True) parent = GFKSerializerField(read_only=True)
image_width = serializers.IntegerField(read_only=True) image_width = serializers.IntegerField(read_only=True)
image_height = serializers.IntegerField(read_only=True) image_height = serializers.IntegerField(read_only=True)
image_size = serializers.IntegerField(read_only=True)
class Meta: class Meta:
model = ImageAttachment model = ImageAttachment
fields = [ fields = [
'id', 'url', 'display', 'object_type', 'object_id', 'parent', 'name', 'image', 'description', 'id', 'url', 'display', 'object_type', 'object_id', 'parent', 'name', 'image', 'description',
'image_height', 'image_width', 'created', 'last_updated', 'image_height', 'image_width', 'image_size', 'created', 'last_updated',
] ]
brief_fields = ('id', 'url', 'display', 'name', 'image', 'description') brief_fields = ('id', 'url', 'display', 'name', 'image', 'description')

View File

@ -506,7 +506,9 @@ class ImageAttachmentFilterSet(ChangeLoggedModelFilterSet):
class Meta: class Meta:
model = ImageAttachment model = ImageAttachment
fields = ('id', 'object_type_id', 'object_id', 'name', 'description', 'image_width', 'image_height') fields = (
'id', 'object_type_id', 'object_id', 'name', 'description', 'image_width', 'image_height', 'image_size',
)
def search(self, queryset, name, value): def search(self, queryset, name, value):
if not value.strip(): if not value.strip():

View File

@ -24,7 +24,14 @@ if TYPE_CHECKING:
) )
from extras.graphql.filter_lookups import ExtraChoicesLookup from extras.graphql.filter_lookups import ExtraChoicesLookup
from netbox.graphql.enums import ColorEnum from netbox.graphql.enums import ColorEnum
from netbox.graphql.filter_lookups import FloatLookup, IntegerLookup, JSONFilter, StringArrayLookup, TreeNodeFilter from netbox.graphql.filter_lookups import (
BigIntegerLookup,
FloatLookup,
IntegerLookup,
JSONFilter,
StringArrayLookup,
TreeNodeFilter,
)
from tenancy.graphql.filters import TenantFilter, TenantGroupFilter from tenancy.graphql.filters import TenantFilter, TenantGroupFilter
from users.graphql.filters import GroupFilter, UserFilter from users.graphql.filters import GroupFilter, UserFilter
from virtualization.graphql.filters import ClusterFilter, ClusterGroupFilter, ClusterTypeFilter from virtualization.graphql.filters import ClusterFilter, ClusterGroupFilter, ClusterTypeFilter
@ -275,6 +282,9 @@ class ImageAttachmentFilter(ChangeLoggedModelFilter):
image_width: Annotated['IntegerLookup', strawberry.lazy('netbox.graphql.filter_lookups')] | None = ( image_width: Annotated['IntegerLookup', strawberry.lazy('netbox.graphql.filter_lookups')] | None = (
strawberry_django.filter_field() strawberry_django.filter_field()
) )
image_size: Annotated['BigIntegerLookup', strawberry.lazy('netbox.graphql.filter_lookups')] | None = (
strawberry_django.filter_field()
)
name: StrFilterLookup | None = strawberry_django.filter_field() name: StrFilterLookup | None = strawberry_django.filter_field()

View File

@ -0,0 +1,62 @@
from django.core.management.base import BaseCommand
from extras.models import ImageAttachment
# Number of records to read and write per batch
BATCH_SIZE = 100
class Command(BaseCommand):
help = "Populate the image_size field for ImageAttachments that predate it"
def handle(self, *args, **options):
verbosity = options['verbosity']
# Only consider attachments whose size has not yet been cached. This is safe to re-run: rows whose file
# was unreadable on a previous run remain NULL and will be retried.
queryset = ImageAttachment.objects.filter(image_size__isnull=True)
total = queryset.count()
if not total:
if verbosity:
self.stdout.write(self.style.SUCCESS("No image attachments require updating."))
return
if verbosity:
self.stdout.write(f"Populating image_size for {total} image attachment(s)...")
updated = 0
skipped = 0
batch = []
for processed, attachment in enumerate(queryset.iterator(chunk_size=BATCH_SIZE), start=1):
# These rows have image_size=NULL, so the size property reads the file from storage (returning None if
# it's inaccessible) rather than returning a cached value.
size = attachment.size
if size is None:
# File is inaccessible; leave image_size NULL so a future run can retry.
skipped += 1
else:
attachment.image_size = size
batch.append(attachment)
if len(batch) >= BATCH_SIZE:
ImageAttachment.objects.bulk_update(batch, ['image_size'])
updated += len(batch)
batch = []
# Reading each file's size may issue a request to the storage backend, so emit periodic progress
# for large tables rather than going silent until completion.
if verbosity and processed % BATCH_SIZE == 0:
self.stdout.write(f" Processed {processed}/{total}...")
if batch:
ImageAttachment.objects.bulk_update(batch, ['image_size'])
updated += len(batch)
if verbosity:
self.stdout.write(self.style.SUCCESS(f"Updated {updated} image attachment(s)."))
if skipped:
self.stdout.write(self.style.WARNING(
f"Skipped {skipped} inaccessible file(s); re-run this command to retry them."
))

View File

@ -0,0 +1,16 @@
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("extras", "0139_alter_customfieldchoiceset_extra_choices"),
]
operations = [
migrations.AddField(
model_name="imageattachment",
name="image_size",
field=models.PositiveBigIntegerField(blank=True, null=True),
),
]

View File

@ -702,6 +702,14 @@ class ImageAttachment(ChangeLoggedModel):
image_width = models.PositiveSmallIntegerField( image_width = models.PositiveSmallIntegerField(
verbose_name=_('image width'), verbose_name=_('image width'),
) )
# Unlike image_height/image_width (populated automatically by ImageField), there is no native size_field, so
# this is populated in save(). It is nullable because existing rows predate the field and storage reads can
# fail; a null value means "not yet computed" and the size property falls back to reading storage.
image_size = models.PositiveBigIntegerField(
verbose_name=_('image size'),
blank=True,
null=True,
)
name = models.CharField( name = models.CharField(
verbose_name=_('name'), verbose_name=_('name'),
max_length=50, max_length=50,
@ -717,6 +725,27 @@ class ImageAttachment(ChangeLoggedModel):
clone_fields = ('object_type', 'object_id') clone_fields = ('object_type', 'object_id')
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Cache an identity for the current image so save() can detect a new/replaced file and recompute the cached
# image_size. We combine the file name with the (auto-populated) dimensions: a replacement that reuses the
# same name is still caught when its dimensions differ. Read the raw image value from __dict__ to avoid
# triggering the ImageField descriptor here (doing so during ORM/GraphQL instantiation can recurse).
self._orig_image_key = self._image_identity()
def _image_identity(self):
"""
Return a tuple identifying the current image file for change detection: its name plus the dimensions Django
populates from it. All three are read raw from __dict__ to avoid triggering the ImageField descriptor
(accessing `self.image` during ORM/GraphQL instantiation can recurse). Not a content fingerprint: a
replacement with an identical name AND identical dimensions is not distinguished (would require reading the
file, the storage round-trip this caching avoids).
"""
original = self.__dict__.get('image')
name = getattr(original, 'name', original)
return (name, self.__dict__.get('image_height'), self.__dict__.get('image_width'))
class Meta: class Meta:
ordering = ('name', 'pk') # name may be non-unique ordering = ('name', 'pk') # name may be non-unique
indexes = ( indexes = (
@ -772,12 +801,15 @@ class ImageAttachment(ChangeLoggedModel):
alt_text=escape(self.description or self.name), alt_text=escape(self.description or self.name),
)) ))
@property def _read_image_size(self):
def size(self):
""" """
Wrapper around `image.size` to suppress an OSError in case the file is inaccessible. Also opportunistically Read the image file's size from storage, suppressing an OSError in case the file is inaccessible. Also
catch other exceptions that we know other storage back-ends to throw. opportunistically catch other exceptions that we know other storage back-ends to throw. Returns None if the
size cannot be determined. This may issue a request to the storage backend (e.g. a HEAD request to S3).
""" """
if not self.image:
return None
expected_exceptions = [OSError] expected_exceptions = [OSError]
try: try:
@ -791,6 +823,33 @@ class ImageAttachment(ChangeLoggedModel):
except tuple(expected_exceptions): except tuple(expected_exceptions):
return None return None
@property
def size(self):
"""
Return the size of the image file in bytes. Prefer the cached `image_size` value to avoid a storage request;
fall back to reading from storage for legacy rows where `image_size` has not yet been populated.
"""
if self.image_size is not None:
return self.image_size
return self._read_image_size()
def save(self, *args, **kwargs):
# Populate image_size on creation or when the image file has changed. Reading the size may touch the storage
# backend (e.g. a HEAD request to S3), so we only do it when necessary: bulk operations that don't alter the
# image (bulk edit, rename) leave the identity unchanged and skip the read entirely. We never overwrite a good
# value with None (e.g. on a transient storage error); a failed read while replacing a file keeps the prior
# size until the next successful save, which is preferred over storing None.
orig_image_key = getattr(self, '_orig_image_key', None)
if self._state.adding or self._image_identity() != orig_image_key:
size = self._read_image_size()
if size is not None:
self.image_size = size
super().save(*args, **kwargs)
# Refresh the cached identity so subsequent saves on this instance detect further changes correctly.
self._orig_image_key = self._image_identity()
def to_objectchange(self, action): def to_objectchange(self, action):
objectchange = super().to_objectchange(action) objectchange = super().to_objectchange(action)
objectchange.related_object = self.parent objectchange.related_object = self.parent

View File

@ -774,7 +774,8 @@ class ImageAttachmentTestCase(
name='Image Attachment 1', name='Image Attachment 1',
image='http://example.com/image1.png', image='http://example.com/image1.png',
image_height=100, image_height=100,
image_width=100 image_width=100,
image_size=1024
), ),
ImageAttachment( ImageAttachment(
object_type=ct, object_type=ct,
@ -782,7 +783,8 @@ class ImageAttachmentTestCase(
name='Image Attachment 2', name='Image Attachment 2',
image='http://example.com/image2.png', image='http://example.com/image2.png',
image_height=100, image_height=100,
image_width=100 image_width=100,
image_size=2048
), ),
ImageAttachment( ImageAttachment(
object_type=ct, object_type=ct,
@ -790,7 +792,8 @@ class ImageAttachmentTestCase(
name='Image Attachment 3', name='Image Attachment 3',
image='http://example.com/image3.png', image='http://example.com/image3.png',
image_height=100, image_height=100,
image_width=100 image_width=100,
image_size=4096
) )
) )
ImageAttachment.objects.bulk_create(image_attachments) ImageAttachment.objects.bulk_create(image_attachments)

View File

@ -741,7 +741,8 @@ class ImageAttachmentTestCase(TestCase, ChangeLoggedFilterSetTests):
name='Image Attachment 1', name='Image Attachment 1',
image='http://example.com/image1.png', image='http://example.com/image1.png',
image_height=100, image_height=100,
image_width=100 image_width=100,
image_size=1024
), ),
ImageAttachment( ImageAttachment(
object_type=site_ct, object_type=site_ct,
@ -749,7 +750,8 @@ class ImageAttachmentTestCase(TestCase, ChangeLoggedFilterSetTests):
name='Image Attachment 2', name='Image Attachment 2',
image='http://example.com/image2.png', image='http://example.com/image2.png',
image_height=100, image_height=100,
image_width=100 image_width=100,
image_size=2048
), ),
ImageAttachment( ImageAttachment(
object_type=rack_ct, object_type=rack_ct,
@ -757,7 +759,8 @@ class ImageAttachmentTestCase(TestCase, ChangeLoggedFilterSetTests):
name='Image Attachment 3', name='Image Attachment 3',
image='http://example.com/image3.png', image='http://example.com/image3.png',
image_height=100, image_height=100,
image_width=100 image_width=100,
image_size=4096
), ),
ImageAttachment( ImageAttachment(
object_type=rack_ct, object_type=rack_ct,
@ -765,7 +768,8 @@ class ImageAttachmentTestCase(TestCase, ChangeLoggedFilterSetTests):
name='Image Attachment 4', name='Image Attachment 4',
image='http://example.com/image4.png', image='http://example.com/image4.png',
image_height=100, image_height=100,
image_width=100 image_width=100,
image_size=8192
) )
) )
ImageAttachment.objects.bulk_create(image_attachments) ImageAttachment.objects.bulk_create(image_attachments)
@ -789,6 +793,17 @@ class ImageAttachmentTestCase(TestCase, ChangeLoggedFilterSetTests):
} }
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1) self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_image_size(self):
# Fixtures set image_size to 1024, 2048, 4096, 8192.
params = {'image_size': [1024, 2048]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
def test_image_size_range(self):
# __gte/__lte bound the 1024/2048/4096/8192 fixtures to the middle two. NetBox's auto-generated numeric
# lookups are multi-value, so values are passed as lists.
params = {'image_size__gte': [2048], 'image_size__lte': [4096]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
class TableConfigTestCase(TestCase, ChangeLoggedFilterSetTests): class TableConfigTestCase(TestCase, ChangeLoggedFilterSetTests):
queryset = TableConfig.objects.all() queryset = TableConfig.objects.all()

View File

@ -11,6 +11,8 @@ from dcim.choices import InterfaceTypeChoices
from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site from dcim.models import Device, DeviceRole, DeviceType, Interface, Manufacturer, Site
from extras.management.commands import renaturalize, webhook_receiver from extras.management.commands import renaturalize, webhook_receiver
from extras.management.commands.webhook_receiver import WebhookHandler from extras.management.commands.webhook_receiver import WebhookHandler
from extras.models import ImageAttachment
from extras.tests.test_models import OverwriteStyleMemoryStorage, UnreadableSizeMemoryStorage
from users.models import User from users.models import User
from utilities.fields import NaturalOrderingField from utilities.fields import NaturalOrderingField
@ -502,3 +504,65 @@ class WebhookReceiverTestCase(TestCase):
print_.assert_any_call('(No body)') print_.assert_any_call('(No body)')
print_.assert_any_call('Completed request #1') print_.assert_any_call('Completed request #1')
class PopulateImageSizesTestCase(TestCase):
@classmethod
def setUpTestData(cls):
cls.ct_site = ContentType.objects.get_by_natural_key('dcim', 'site')
cls.site = Site.objects.create(name='Site 1', slug='site-1')
def _legacy_attachment(self, name, image_name):
# bulk_create skips save(), so image_size starts NULL, mimicking a row created before the field existed.
attachment = ImageAttachment(
object_type=self.ct_site,
object_id=self.site.pk,
name=name,
image=image_name,
image_height=100,
image_width=100,
)
ImageAttachment.objects.bulk_create([attachment])
return ImageAttachment.objects.get(name=name)
def test_populates_null_image_sizes(self):
field = ImageAttachment._meta.get_field('image')
storage = OverwriteStyleMemoryStorage()
storage.files['image-attachments/site_1_a.png'] = b'\x00' * 321
attachment = self._legacy_attachment('Legacy 1', 'image-attachments/site_1_a.png')
self.assertIsNone(attachment.image_size)
out = StringIO()
with patch.object(field, 'storage', storage):
call_command('populate_image_sizes', stdout=out)
# Read the persisted value back under the patched storage (refreshing the image field re-reads
# dimensions from storage, which must be the in-memory backend).
persisted = ImageAttachment.objects.get(pk=attachment.pk)
self.assertEqual(persisted.image_size, 321)
self.assertIn('Updated 1', out.getvalue())
def test_skips_unreadable_files_and_is_rerunnable(self):
field = ImageAttachment._meta.get_field('image')
attachment = self._legacy_attachment('Legacy 2', 'image-attachments/site_1_missing.png')
# First run: storage can't report size, so the row is skipped and left NULL.
out = StringIO()
with patch.object(field, 'storage', UnreadableSizeMemoryStorage()):
call_command('populate_image_sizes', stdout=out)
self.assertIsNone(ImageAttachment.objects.get(pk=attachment.pk).image_size)
self.assertIn('Skipped 1', out.getvalue())
# Second run: storage now works, and the previously-skipped row is picked up.
storage = OverwriteStyleMemoryStorage()
storage.files['image-attachments/site_1_missing.png'] = b'\x00' * 654
out = StringIO()
with patch.object(field, 'storage', storage):
call_command('populate_image_sizes', stdout=out)
self.assertEqual(ImageAttachment.objects.get(pk=attachment.pk).image_size, 654)
def test_noop_when_nothing_to_update(self):
out = StringIO()
call_command('populate_image_sizes', stdout=out)
self.assertIn('No image attachments require updating.', out.getvalue())

View File

@ -74,6 +74,16 @@ class OverwriteStyleMemoryStorage(Storage):
return f'https://example.invalid/{name}' return f'https://example.invalid/{name}'
class UnreadableSizeMemoryStorage(OverwriteStyleMemoryStorage):
"""
Like OverwriteStyleMemoryStorage, but size() raises OSError to model a storage backend that is
transiently unavailable (e.g. an S3 outage) when reading file size.
"""
def size(self, name):
raise OSError('storage unavailable')
class ImageAttachmentTestCase(TestCase): class ImageAttachmentTestCase(TestCase):
@classmethod @classmethod
def setUpTestData(cls): def setUpTestData(cls):
@ -191,6 +201,204 @@ class ImageAttachmentTestCase(TestCase):
self.assertCountEqual(storage.files.keys(), {base_name, suffixed_name}) self.assertCountEqual(storage.files.keys(), {base_name, suffixed_name})
def test_save_populates_image_size_on_create(self):
"""
save() populates image_size from the uploaded file on creation.
"""
storage = OverwriteStyleMemoryStorage()
field = ImageAttachment._meta.get_field('image')
with patch.object(field, 'storage', storage):
ia = ImageAttachment(
object_type=self.ct_site,
object_id=self.site.pk,
image=self._uploaded_png('size-on-create.png'),
)
ia.save()
self.assertIsNotNone(ia.image_size)
self.assertEqual(ia.image_size, ia.image.size)
def test_size_property_returns_stored_value_without_storage_access(self):
"""
The size property returns the cached image_size rather than the file's actual size. The stub's empty
file reports size 0, so asserting the distinct stored value proves the property used the cached value.
"""
ia = self._stub_image_attachment(self.site.pk, 'image-attachments/site_1_no-file.png')
self.assertEqual(ia._read_image_size(), 0) # the stub's empty file genuinely reports 0
ia.image_size = 9999
self.assertEqual(ia.size, 9999)
def test_size_property_falls_back_to_storage_when_unset(self):
"""
For legacy rows where image_size is NULL, the size property falls back to reading storage
(rather than reporting 0 bytes).
"""
storage = OverwriteStyleMemoryStorage()
field = ImageAttachment._meta.get_field('image')
with patch.object(field, 'storage', storage):
ia = ImageAttachment(
object_type=self.ct_site,
object_id=self.site.pk,
image=self._uploaded_png('fallback.png'),
)
ia.save()
# Simulate a legacy row that predates the image_size field.
ia.image_size = None
self.assertEqual(ia.size, ia.image.size)
self.assertGreater(ia.size, 0)
def test_save_does_not_clobber_existing_size_on_storage_error(self):
"""
When the storage backend raises on a size read (modeled by a real Storage subclass, not a mock),
save() must not overwrite an existing image_size with None.
"""
field = ImageAttachment._meta.get_field('image')
# Create a row with a real, readable size.
with patch.object(field, 'storage', OverwriteStyleMemoryStorage()):
ia = ImageAttachment(
object_type=self.ct_site,
object_id=self.site.pk,
image=self._uploaded_png('keep-size.png'),
)
ia.save()
original_size = ia.image_size
self.assertIsNotNone(original_size)
# Reload from the DB so the FieldFile has no cached size and must consult storage (as it would for a
# row loaded fresh in production). With the backend unable to report size, the read fails (returns None),
# and save() must keep the previously-stored value rather than clobbering it with None.
with patch.object(field, 'storage', UnreadableSizeMemoryStorage()):
reloaded = ImageAttachment.objects.get(pk=ia.pk)
self.assertIsNone(reloaded._read_image_size()) # the read genuinely fails (returns None)
# Make the image look replaced by perturbing the cached identity (different name component).
reloaded._orig_image_key = ('image-attachments/site_1_old.png', reloaded.image_height, reloaded.image_width)
reloaded.save()
# In-memory value is preserved, and the persisted value is unchanged.
self.assertEqual(reloaded.image_size, original_size)
self.assertEqual(ImageAttachment.objects.get(pk=ia.pk).image_size, original_size)
def test_save_recomputes_image_size_when_image_replaced(self):
"""
Replacing the image on an existing row recomputes image_size (Cable-style change detection).
"""
storage = OverwriteStyleMemoryStorage()
field = ImageAttachment._meta.get_field('image')
with patch.object(field, 'storage', storage):
ia = ImageAttachment(
object_type=self.ct_site,
object_id=self.site.pk,
image=self._uploaded_png('original.png'),
)
ia.save()
original_size = ia.image_size
self.assertIsNotNone(original_size)
# Replace the image with a larger file and save again.
larger = SimpleUploadedFile(
name='replacement.png',
content=self._uploaded_png('replacement.png').read() + b'\x00' * 100,
content_type='image/png',
)
ia.image = larger
ia.save()
self.assertEqual(ia.image_size, ia.image.size)
self.assertNotEqual(ia.image_size, original_size)
def test_image_identity_includes_dimensions(self):
"""
The change-detection key combines the image name with its dimensions, so a replacement that reuses the
same name but changes dimensions produces a different key (which name alone would not).
"""
ia = self._stub_image_attachment(self.site.pk, 'image-attachments/site_1_same.png')
ia.image_height, ia.image_width = 10, 10
key_small = ia._image_identity()
# Same name, different dimensions (as Django would set when a same-named file is replaced).
ia.image_height, ia.image_width = 40, 40
key_large = ia._image_identity()
self.assertEqual(key_small[0], key_large[0]) # name component unchanged
self.assertNotEqual(key_small, key_large) # but the key differs, so save() will recompute
def test_save_recomputes_image_size_when_dimensions_change_under_same_name(self):
"""
When the image is replaced by a file with the same stored name but different dimensions, save()
recomputes image_size. Name-only detection would miss this; the dimension component catches it.
Simulates the same-name case by priming the cached identity with the old dimensions.
"""
storage = OverwriteStyleMemoryStorage()
field = ImageAttachment._meta.get_field('image')
with patch.object(field, 'storage', storage):
ia = ImageAttachment(
object_type=self.ct_site,
object_id=self.site.pk,
image=self._uploaded_png('same-name.png'),
)
ia.save()
name = ia.image.name
# Force the cached identity to reflect the SAME name but different (old) dimensions, then bump the
# current dimensions to mimic a same-name replacement with a differently-sized image.
ia._orig_image_key = (name, ia.image_height + 5, ia.image_width + 5)
ia.save()
self.assertEqual(ia.image.name, name) # name unchanged
self.assertEqual(ia.image_size, ia.image.size) # size recomputed despite same name
def test_save_without_touching_image_does_not_recompute_or_read_storage(self):
"""
Editing an existing row without replacing the image leaves image_size untouched and does not
hit storage. Directly guards against the change-detection comparison misfiring.
"""
storage = OverwriteStyleMemoryStorage()
field = ImageAttachment._meta.get_field('image')
with patch.object(field, 'storage', storage):
ia = ImageAttachment(
object_type=self.ct_site,
object_id=self.site.pk,
name='Original',
image=self._uploaded_png('untouched.png'),
)
ia.save()
stored_size = ia.image_size
# Reload from the DB so the cached image identity is set from the persisted value, then edit only the name.
reloaded = ImageAttachment.objects.get(pk=ia.pk)
reloaded.name = 'Renamed'
with patch.object(ImageAttachment, '_read_image_size', side_effect=AssertionError('storage accessed')):
reloaded.save()
self.assertEqual(reloaded.image_size, stored_size)
def test_save_populates_image_size_via_constructor_kwarg(self):
"""
The non-UI create path (constructor kwarg / REST / bulk) populates image_size correctly,
confirming change detection behaves when image is passed as a FieldFile.
"""
storage = OverwriteStyleMemoryStorage()
field = ImageAttachment._meta.get_field('image')
with patch.object(field, 'storage', storage):
ia = ImageAttachment(
object_type=self.ct_site,
object_id=self.site.pk,
image=self._uploaded_png('kwarg.png'),
)
ia.save()
self.assertIsNotNone(ia.image_size)
self.assertEqual(ia.image_size, ia.image.size)
class TableConfigTestCase(TestCase): class TableConfigTestCase(TestCase):
@classmethod @classmethod

View File

@ -481,6 +481,7 @@ class ImageAttachmentTestCase(
image='http://example.com/image1.png', image='http://example.com/image1.png',
image_height=100, image_height=100,
image_width=100, image_width=100,
image_size=1024,
), ),
ImageAttachment( ImageAttachment(
object_type=ct, object_type=ct,
@ -489,6 +490,7 @@ class ImageAttachmentTestCase(
image='http://example.com/image2.png', image='http://example.com/image2.png',
image_height=100, image_height=100,
image_width=100, image_width=100,
image_size=2048,
), ),
ImageAttachment( ImageAttachment(
object_type=ct, object_type=ct,
@ -497,6 +499,7 @@ class ImageAttachmentTestCase(
image='http://example.com/image3.png', image='http://example.com/image3.png',
image_height=100, image_height=100,
image_width=100, image_width=100,
image_size=4096,
), ),
] ]
) )