gru/compare_filesize_drupal6
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixes

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-10-23 15:26:08 +02:00
1 parent bb941cff43
commit 6a821c7512
7 files changed
+1768 -326
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_drupal6.py
+379
View File
@@ -0,0 +1,379 @@
import io
import os
import re
import sys
import tempfile
import types
import unittest
from contextlib import redirect_stdout
from pathlib import Path
from unittest.mock import patch
from PIL import Image
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
class FakeCursor:
def __init__(self, conn):
self.conn = conn
self.results = []
self.index = 0
self.rowcount = 0
def __enter__(self):
return self
def __exit__(self, *_):
return False
def close(self):
pass
def execute(self, sql, params=None):
params = params or ()
self.index = 0
self.rowcount = 0
self.results = []
if sql == 'SHOW TABLES':
self.results = [{'Tables_in_test': table} for table in self.conn.schemas]
elif sql.startswith('SHOW COLUMNS FROM'):
table = re.search(r'`([^`]+)`', sql).group(1)
self.results = [{'Field': name, 'Type': data_type} for name, data_type in self.conn.schemas[table].items()]
elif sql.startswith('SELECT fid, filepath, filesize FROM'):
self.results = [dict(row) for row in self.conn.files]
elif sql.startswith('SELECT filesize, filepath FROM'):
self.results = [dict(row) for row in self.conn.files if row['fid'] == params[0]]
elif sql.startswith('UPDATE `files` SET filesize'):
if self.conn.fail_updates:
self.conn.fail_updates -= 1
raise RuntimeError('Simulated DB error')
size, fid, path = params
for row in self.conn.files:
if row['fid'] == fid and row['filepath'] == path:
if row['filesize'] != size:
row['filesize'] = size
self.rowcount = 1
break
return
elif 'FROM `upload` t JOIN' in sql or 'FROM `content_field_scan` t JOIN' in sql:
table = 'upload' if 'FROM `upload`' in sql else 'content_field_scan'
field = 'fid' if table == 'upload' else 'field_scan_fid'
for entry in self.conn.relations.get(table, []):
node = self.conn.nodes.get(entry['nid'])
if not node or (entry.get('vid') and node['vid'] != entry['vid']):
continue
if entry[field] in params:
self.results.append({
'nid': node['nid'], 'title': node['title'],
'node_type': node['type'], 'node_status': node['status'],
'fid': entry[field], 'cid': entry.get('cid'),
})
elif 'FROM `comment_upload` t' in sql:
for entry in self.conn.relations.get('comment_upload', []):
comment = next(c for c in self.conn.comments if c['cid'] == entry['cid'])
node = self.conn.nodes[comment['nid']]
if entry['fid'] in params:
self.results.append({'nid': node['nid'], 'title': node['title'],
'node_type': node['type'], 'node_status': node['status'],
'comment_status': comment['status'],
'cid': comment['cid'], 'fid': entry['fid']})
elif 'r.body AS body' in sql:
for node in self.conn.nodes.values():
for revision in self.conn.revisions:
if revision['nid'] == node['nid'] and revision['vid'] == node['vid']:
self.results.append({
'nid': node['nid'], 'title': node['title'], 'node_type': node['type'],
'node_status': node['status'], 'body': revision['body'],
})
elif 'c.cid, c.status AS comment_status' in sql:
for comment in self.conn.comments:
node = self.conn.nodes[comment['nid']]
self.results.append({
'nid': node['nid'], 'title': node['title'], 'node_type': node['type'],
'node_status': node['status'], 'cid': comment['cid'],
'comment_status': comment['status'], 'comment': comment['comment'],
})
else:
raise AssertionError(f'Unexpected fake SQL: {sql} / {params}')
self.rowcount = len(self.results)
def fetchall(self):
return list(self.results)
def fetchone(self):
return self.results[0] if self.results else None
def fetchmany(self, count):
result = self.results[self.index:self.index + count]
self.index += len(result)
return result
class FakeConnection:
def __init__(self):
self.schemas = {
'files': {'fid': 'int', 'filepath': 'varchar(255)', 'filesize': 'int'},
'node': {'nid': 'int', 'vid': 'int', 'title': 'varchar(255)', 'type': 'varchar(32)', 'status': 'tinyint'},
'node_revisions': {'nid': 'int', 'vid': 'int', 'body': 'longtext'},
'upload': {'fid': 'int', 'nid': 'int', 'vid': 'int'},
'content_field_scan': {'field_scan_fid': 'int', 'nid': 'int', 'vid': 'int'},
'comments': {'cid': 'int', 'nid': 'int', 'comment': 'longtext', 'status': 'tinyint'},
'comment_upload': {'cid': 'int', 'fid': 'int'},
}
self.nodes = {
11: {'nid': 11, 'vid': 101, 'title': 'Stary skan mapy', 'type': 'forum', 'status': 1},
12: {'nid': 12, 'vid': 102, 'title': 'Zdjecia z wakacji', 'type': 'story', 'status': 1},
13: {'nid': 13, 'vid': 103, 'title': 'Odpowiedz na forum', 'type': 'forum', 'status': 0},
}
self.relations = {
'upload': [{'nid': 11, 'vid': 101, 'fid': 7}, {'nid': 13, 'vid': 999, 'fid': 7}],
'content_field_scan': [{'nid': 12, 'vid': 102, 'field_scan_fid': 8}],
'comment_upload': [{'cid': 55, 'fid': 8}],
}
self.revisions = [
{'nid': 11, 'vid': 101, 'body': '<img src="/sites/default/files/scans/mapa.jpg">'},
{'nid': 12, 'vid': 102, 'body': '<p>Wspomnienia.</p>'},
{'nid': 13, 'vid': 103, 'body': 'Zobacz mapa.jpg tutaj.'},
{'nid': 13, 'vid': 999, 'body': 'NIEAKTUALNE /sites/default/files/scans/mapa.jpg'},
]
self.comments = [
{'cid': 55, 'nid': 12, 'comment': 'Znalazlem /sites/default/files/scans/mapa.jpg', 'status': 1},
]
self.files = []
self.fail_updates = 0
self.commits = 0
self.rollbacks = 0
def __enter__(self):
return self
def __exit__(self, *_):
return False
def cursor(self, *_):
return FakeCursor(self)
def commit(self):
self.commits += 1
def rollback(self):
self.rollbacks += 1
class TestLinks(unittest.TestCase):
def test_links(self):
self.assertEqual(node_url('https://example.org/forum/', 42), 'https://example.org/forum/node/42')
self.assertEqual(node_url('https://example.org/forum', 42, 66), 'https://example.org/forum/node/42#comment-66')
self.assertEqual(node_url('https://example.org/forum', 42, clean_urls=False), 'https://example.org/forum/?q=node/42')
self.assertIn('skan%20mapa.jpg', search_url('https://example.org', 'skan mapa.jpg'))
self.assertEqual(node_url('https://example.org', 42, 66, comment_permalinks=True),
'https://example.org/comment/66#comment-66')
def test_matcher_precision(self):
matcher = ReferenceMatcher([
{'fid': 7, 'db_filepath': 'sites/default/files/scans/mapa.jpg', 'rel_path': 'scans/mapa.jpg'},
])
self.assertEqual(matcher.match('<img src="/sites/default/files/scans/mapa.jpg">')[7], 'sciezka')
self.assertEqual(matcher.match('mapa.jpg')[7], 'nazwa')
self.assertFalse(matcher.match('supermapa.jpg'))
self.assertFalse(ReferenceMatcher([
{'fid': 7, 'db_filepath': 'sites/default/files/scans/mapa.jpg', 'rel_path': 'scans/mapa.jpg'}
], basename_match=False).match('mapa.jpg'))
def test_schema_references_current_only_and_comment_status(self):
conn = FakeConnection()
lookup = DrupalReferenceLookup(conn, base_url='https://forum.example.org')
files = [
{'fid': 7, 'db_filepath': 'sites/default/files/scans/mapa.jpg', 'rel_path': 'scans/mapa.jpg'},
{'fid': 8, 'db_filepath': 'sites/default/files/inne.jpg', 'rel_path': 'inne.jpg'},
]
fake_pymysql = types.SimpleNamespace(cursors=types.SimpleNamespace(SSDictCursor=object))
with patch.dict(sys.modules, {'pymysql': fake_pymysql}):
with redirect_stdout(io.StringIO()):
result = lookup.lookup(files)
self.assertEqual({x['nid'] for x in result[7]}, {11, 12, 13})
self.assertEqual({x['nid'] for x in result[8]}, {12})
attached_to_comment = [x for x in result[8] if x['cid'] == 55][0]
self.assertIn('comment_upload.fid', attached_to_comment['sources'])
direct = [x for x in result[7] if x['nid'] == 11][0]
self.assertIn('upload.fid', direct['sources'])
self.assertIn('tresc wpisu (sciezka)', direct['sources'])
self.assertEqual(direct['confidence'], 'pewne')
self.assertTrue(direct['url'].endswith('/node/11'))
comment = [x for x in result[7] if x['cid'] == 55][0]
self.assertFalse(comment['comment_published'])
self.assertEqual(comment['url'], 'https://forum.example.org/node/12#comment-55')
possible = [x for x in result[7] if x['nid'] == 13][0]
self.assertFalse(possible['published'])
self.assertEqual(possible['confidence'], 'mozliwe')
self.assertNotIn('upload.fid', possible['sources']) # nieaktualna rewizja
class TestFiles(unittest.TestCase):
def setUp(self):
self.workdir = tempfile.TemporaryDirectory()
self.addCleanup(self.workdir.cleanup)
self.root = Path(self.workdir.name)
self.site_root = self.root / 'drupal'
self.folder = self.site_root / 'sites' / 'default' / 'files'
(self.folder / 'scans').mkdir(parents=True)
self.image = self.folder / 'scans' / 'mapa.jpg'
src = Image.effect_noise((950, 1250), 115).convert('RGB')
src.save(self.image, 'JPEG', quality=99)
self.conn = FakeConnection()
self.conn.files = [{
'fid': 7, 'filepath': 'sites/default/files/scans/mapa.jpg', 'filesize': 12345,
}]
self.cfg = types.SimpleNamespace(
FILES_BASE_PATH=str(self.folder), PUBLIC_FILES_PATH='sites/default/files',
DRUPAL_ROOT=str(self.site_root), DRUPAL_BASE_URL='https://forum.example.org',
)
def args(self, *overrides):
return main.build_parser().parse_args(list(overrides))
def test_resolve_prevent_traversal(self):
valid, rel = main.resolve_filepath('sites/default/files/scans/mapa.jpg', str(self.folder))
self.assertEqual(valid, str(self.image))
self.assertEqual(rel, 'scans/mapa.jpg')
bad, _ = main.resolve_filepath('../../etc/shadow', str(self.folder))
self.assertIsNone(bad)
outside, _ = main.resolve_filepath('/etc/passwd', str(self.folder))
self.assertIsNone(outside)
def test_backup_must_not_be_in_webroot(self):
with self.assertRaises(ValueError):
main._safe_backup_dir(str(self.site_root / 'backup'),
str(self.folder), str(self.site_root))
def test_dry_run_no_changes_and_preview(self):
args = self.args('--optimize', '--update-db', '--dry-run', '--max-width', '600',
'--max-height', '800', '--max-file-mb', '0.5')
images, missing, prefix = main.collect_images(self.conn, self.cfg, args)
self.assertFalse(missing)
before = self.image.read_bytes()
with redirect_stdout(io.StringIO()):
stats = main.process_changes(self.conn, self.cfg, args, images, prefix)
self.assertEqual(self.image.read_bytes(), before)
self.assertEqual(self.conn.files[0]['filesize'], 12345)
self.assertEqual(stats['optimized'], 0)
self.assertEqual(stats['planned'], 1)
self.assertEqual(images[0]['optimize_status'], 'planned')
self.assertLess(images[0]['after_size'], images[0]['before_size'])
def test_apply_backup_db_and_restore_on_sql_failure(self):
args = self.args('--optimize', '--apply', '--max-width', '600', '--max-height', '800',
'--max-file-mb', '0.5', '--backup-dir', str(self.root / 'backups'))
before = self.image.read_bytes()
images, _, prefix = main.collect_images(self.conn, self.cfg, args)
with redirect_stdout(io.StringIO()):
stats = main.process_changes(self.conn, self.cfg, args, images, prefix)
self.assertEqual(stats['optimized'], 1)
self.assertEqual(stats['errors'], 0)
self.assertLess(self.image.stat().st_size, len(before))
self.assertEqual(self.conn.files[0]['filesize'], self.image.stat().st_size)
self.assertEqual(Path(images[0]['backup_file']).read_bytes(), before)
with Image.open(self.image) as img:
self.assertLessEqual(img.width, 600)
self.assertLessEqual(img.height, 800)
# Ponownie uruchomienie z bladem w DB: plik i metadata wracaja.
self.conn.fail_updates = 1
self.conn.files[0]['filesize'] = 12345
second_before = self.image.read_bytes()
args2 = self.args('--optimize', '--apply', '--max-width', '400', '--max-height', '550',
'--max-file-mb', '0.3', '--backup-dir', str(self.root / 'backups'))
images2, _, prefix = main.collect_images(self.conn, self.cfg, args2)
with redirect_stdout(io.StringIO()):
stats2 = main.process_changes(self.conn, self.cfg, args2, images2, prefix)
self.assertEqual(stats2['errors'], 1)
self.assertEqual(self.image.read_bytes(), second_before)
self.assertEqual(self.conn.files[0]['filesize'], 12345)
def test_symlink_is_not_modified(self):
link = self.folder / 'scans' / 'alias.jpg'
link.symlink_to(self.image)
self.conn.files[0]['filepath'] = 'sites/default/files/scans/alias.jpg'
args = self.args('--optimize', '--dry-run', '--max-width', '500')
images, _, prefix = main.collect_images(self.conn, self.cfg, args)
self.assertEqual(images[0]['path'], str(link))
with redirect_stdout(io.StringIO()):
stats = main.process_changes(self.conn, self.cfg, args, images, prefix)
self.assertEqual(images[0]['optimize_status'], 'skip')
self.assertEqual(stats['planned'], 0)
self.assertTrue(link.is_symlink())
def test_exif_orientation_after_optimization(self):
original = self.folder / 'scan-orientation.jpg'
img = Image.new('RGB', (240, 120), 'orange')
exif = Image.Exif()
exif[274] = 6
img.save(original, 'JPEG', quality=100, exif=exif)
optimized = create_optimized_candidate(str(original), max_width=150, max_height=200,
max_bytes=100, min_gain_percent=0)
self.assertEqual(optimized['status'], 'ready', optimized['reason'])
try:
with Image.open(optimized['temp_path']) as out:
self.assertEqual(out.size, (100, 200))
self.assertNotEqual(out.getexif().get(274), 6)
finally:
os.unlink(optimized['temp_path'])
def test_png_with_alpha_preserved(self):
filename = self.folder / 'scans' / 'alpha.png'
img = Image.new('RGBA', (300, 900), (10, 100, 50, 130))
img.save(filename)
result = create_optimized_candidate(str(filename), max_width=200, max_height=500,
max_bytes=200, min_gain_percent=0)
self.assertEqual(result['status'], 'ready', result['reason'])
try:
with Image.open(result['temp_path']) as output:
self.assertEqual(output.mode, 'RGBA')
self.assertEqual(output.size, (167, 500))
self.assertEqual(output.getpixel((1, 1))[3], 130)
finally:
os.unlink(result['temp_path'])
def test_no_optimization_for_small_or_gif(self):
tiny = self.folder / 'tiny.png'
Image.new('RGB', (30, 30), 'red').save(tiny)
self.assertEqual(create_optimized_candidate(str(tiny))['status'], 'skip')
gif = self.folder / 'animated.gif'
Image.new('RGB', (300, 500), 'red').save(gif)
self.assertEqual(create_optimized_candidate(str(gif), max_width=100)['status'], 'skip')
def test_cli_end_to_end_read_only_generates_reports(self):
csv_path = self.root / 'end.csv'
json_path = self.root / 'end.json'
with patch.object(main, 'load_config', return_value=self.cfg), \
patch.object(main, 'get_connection', return_value=self.conn), \
redirect_stdout(io.StringIO()):
status = main.main(['--find-usage', '--no-text-scan', '--top', '10',
'--report-csv', str(csv_path),
'--report-json', str(json_path)])
self.assertEqual(status, 0)
self.assertIn('https://forum.example.org/node/11', csv_path.read_text(encoding='utf-8-sig'))
self.assertIn('"images"', json_path.read_text(encoding='utf-8'))
self.assertEqual(self.conn.files[0]['filesize'], 12345)
def test_csv_has_direct_link(self):
args = self.args('--find-usage', '--site-url', 'https://forum.example.org', '--report-csv', str(self.root / 'out.csv'))
images, _, prefix = main.collect_images(self.conn, self.cfg, args)
images[0]['references'] = [
{'nid': 11, 'cid': None, 'title': 'Przypiety temat', 'node_type': 'forum',
'published': True, 'comment_published': None, 'sources': ['upload.fid'],
'confidence': 'pewne', 'url': 'https://forum.example.org/node/11',
'edit_url': 'https://forum.example.org/node/11/edit'}
]
main.write_csv(args.report_csv, images, self.cfg, args)
text = Path(args.report_csv).read_text(encoding='utf-8-sig')
self.assertIn('https://forum.example.org/node/11', text)
self.assertIn('Przypiety temat', text)
if __name__ == '__main__':
unittest.main()