fixes

tests/test_drupal6.py

@@ -1,10 +1,13 @@
 import io
 import os
 import re
+import struct
 import sys
 import tempfile
 import types
 import unittest
+import warnings
+import zlib
 from contextlib import redirect_stdout
 from pathlib import Path
 from unittest.mock import patch
@@ -15,6 +18,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 import main
 from drupal_lookup import DrupalReferenceLookup, ReferenceMatcher, node_url, search_url
 from image_optimizer import create_optimized_candidate
+from image_safety import safe_image_open
 
 
 class FakeCursor:
@@ -346,6 +350,68 @@ class TestFiles(unittest.TestCase):
         Image.new('RGB', (300, 500), 'red').save(gif)
         self.assertEqual(create_optimized_candidate(str(gif), max_width=100)['status'], 'skip')
 
+    def test_image_limit_replaces_pillow_warning_without_disabling_safety(self):
+        # Symulujemy niski prog Pillow: nasz konfigurowalny limit nie
+        # powinien powodowac ostrzezen ani wychodzic poza scope.
+        with patch.object(Image, 'MAX_IMAGE_PIXELS', 100_000):
+            with warnings.catch_warnings(record=True) as emitted:
+                warnings.simplefilter('always', Image.DecompressionBombWarning)
+                self.assertEqual(main.file_resolution(str(self.image), max_pixels=2_000_000),
+                                 (950, 1250))
+                result = create_optimized_candidate(
+                    str(self.image), max_width=550, max_height=700,
+                    max_bytes=100_000, max_pixels=2_000_000,
+                )
+                self.assertEqual(result['status'], 'ready', result['reason'])
+                if result['temp_path']:
+                    os.unlink(result['temp_path'])
+            self.assertEqual(Image.MAX_IMAGE_PIXELS, 100_000)
+            self.assertFalse(any(issubclass(item.category, Image.DecompressionBombWarning)
+                                 for item in emitted))
+
+    def test_giant_image_header_is_readable_but_skipped_above_limit(self):
+        # Header PNG deklaruje tyle pikseli co trzy problematyczne pliki
+        # z logu: 14400 * 13872 = 199 756 800. Nie generujemy tych pikseli.
+        fake = self.folder / 'scans' / '200mp.png'
+        with io.BytesIO() as buffer:
+            Image.new('RGB', (2, 2), 'red').save(buffer, format='PNG')
+            content = bytearray(buffer.getvalue())
+        content[16:24] = struct.pack('>II', 14400, 13872)
+        content[29:33] = struct.pack('>I', zlib.crc32(content[12:29]))
+        fake.write_bytes(content)
+        with patch.object(Image, 'MAX_IMAGE_PIXELS', 100_000):
+            with warnings.catch_warnings(record=True) as emitted:
+                warnings.simplefilter('always', Image.DecompressionBombWarning)
+                self.assertEqual(main.file_resolution(str(fake), max_pixels=220_000_000),
+                                 (14400, 13872))
+                with safe_image_open(str(fake), max_pixels=220_000_000) as source:
+                    self.assertEqual(source.width * source.height, 199_756_800)
+                # Obrazu nie nalezy dekodowac bez limitu RAM.
+                rejected = create_optimized_candidate(str(fake), max_pixels=150_000_000)
+                self.assertEqual(rejected['status'], 'skip')
+                self.assertIn('--max-image-mpixels=150', rejected['reason'])
+                # Przypadek > 2 * limitu Pillow rowniez jest SKIP, nie ERROR.
+                rejected_strict = create_optimized_candidate(str(fake), max_pixels=60_000_000)
+                self.assertEqual(rejected_strict['status'], 'skip')
+                self.assertIn('--max-image-mpixels=60', rejected_strict['reason'])
+            self.assertEqual(Image.MAX_IMAGE_PIXELS, 100_000)
+            self.assertFalse(any(issubclass(item.category, Image.DecompressionBombWarning)
+                                 for item in emitted))
+
+    def test_oversize_images_are_reported_as_skipped_not_errors(self):
+        args = self.args('--optimize', '--dry-run', '--max-image-mpixels', '0.5',
+                         '--max-width', '500')
+        original = self.image.read_bytes()
+        images, _, prefix = main.collect_images(self.conn, self.cfg, args)
+        with redirect_stdout(io.StringIO()) as output:
+            stats = main.process_changes(self.conn, self.cfg, args, images, prefix)
+        self.assertEqual(stats['errors'], 0)
+        self.assertEqual(stats['planned'], 0)
+        self.assertEqual(images[0]['optimize_status'], 'skip')
+        self.assertIn('[SKIP-LIMIT]', output.getvalue())
+        self.assertEqual(self.image.read_bytes(), original)
+        self.assertEqual(self.conn.files[0]['filesize'], 12345)
+
     def test_cli_end_to_end_read_only_generates_reports(self):
         csv_path = self.root / 'end.csv'
         json_path = self.root / 'end.json'