mirror of
https://github.com/inventree/InvenTree.git
synced 2026-10-14 14:11:55 +00:00
* fix(report): support S3 cloud media storage for report and label template loading (#12426) - Prevent TemplateDoesNotExist when using cloud storage backends (e.g. S3 via django-storages) - Update TemplateUploadMixin.template_name to return relative storage path when local file does not exist on disk - Implement ReportTemplateBase.get_template_content() to load content from storage backend and resolve bucket/location prefix mismatches - Add fallback in ReportTemplateBase.render_as_string() to render from storage content when template loader lookup fails - Register InvenTreeStorageTemplateLoader to allow template snippets to resolve directly from default_storage - Add unit tests in S3ReportTemplateTest covering prefix resolution, remote rendering, and storage template loading * fix style * remove weasyprint unavailable hacks this is unrealstic in a full enviroment; the only reason for this is some very constraint sandbox - which is not our run enviroment * fix import --------- Co-authored-by: Matthias Mair <code@mjmair.com> Co-authored-by: Oliver <oliver.henry.walters@gmail.com>
This commit is contained in:
5 files changed
+296
-7
No files matched your search
@@ -139,7 +139,9 @@ def _introspectCommands(argv: list[str]):
|
||||
}
|
||||
|
||||
# The entrypoint itself can indicate the context
|
||||
_context['pytest_entrypoint'] = entrypoint.endswith('pytest')
|
||||
_context['pytest_entrypoint'] = (
|
||||
entrypoint.endswith('pytest') or 'pytest' in entrypoint.lower()
|
||||
)
|
||||
_context['gunicorn_entrypoint'] = 'gunicorn' in entrypoint
|
||||
|
||||
# The development server is running without the auto-reloader
|
||||
@@ -329,7 +331,7 @@ def canAppAccessDatabase(
|
||||
return False
|
||||
|
||||
# Override for testing mode?
|
||||
return allow_test or not cmd_context['test_command']
|
||||
return allow_test or not isInTestMode()
|
||||
|
||||
|
||||
def isPluginRegistryLoaded():
|
||||
|
||||
@@ -502,6 +502,7 @@ TEMPLATES = [
|
||||
'plugin.template.PluginTemplateLoader',
|
||||
'django.template.loaders.filesystem.Loader',
|
||||
'django.template.loaders.app_directories.Loader',
|
||||
'InvenTree.template.InvenTreeStorageTemplateLoader',
|
||||
],
|
||||
)
|
||||
],
|
||||
|
||||
@@ -3,10 +3,66 @@
|
||||
import os
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.files.storage import default_storage
|
||||
from django.template import Origin, TemplateDoesNotExist
|
||||
from django.template.loaders.base import Loader as BaseLoader
|
||||
from django.template.loaders.cached import Loader as CachedLoader
|
||||
|
||||
|
||||
class InvenTreeStorageTemplateLoader(BaseLoader):
|
||||
"""Custom template loader that loads templates from Django's default storage backend (e.g. S3, SFTP)."""
|
||||
|
||||
def get_storage_candidates(self, template_name: str) -> list[str]:
|
||||
"""Generate candidate storage paths for a given template name."""
|
||||
clean = str(template_name).replace('\\', '/')
|
||||
|
||||
# If path starts with MEDIA_ROOT, strip it
|
||||
media_root_str = str(settings.MEDIA_ROOT).replace('\\', '/').rstrip('/')
|
||||
if clean.startswith(media_root_str):
|
||||
clean = clean[len(media_root_str) :].lstrip('/')
|
||||
|
||||
clean = clean.lstrip('/')
|
||||
candidates = [clean]
|
||||
|
||||
# Handle storage location prefix (e.g. S3 / SFTP location)
|
||||
location = getattr(default_storage, 'location', '') or ''
|
||||
location = str(location).replace('\\', '/').strip('/')
|
||||
if location:
|
||||
if clean.startswith(f'{location}/'):
|
||||
candidates.append(clean[len(location) + 1 :])
|
||||
else:
|
||||
candidates.append(f'{location}/{clean}')
|
||||
|
||||
# Handle report subdirectory prefix
|
||||
if not clean.startswith('report/'):
|
||||
candidates.append(f'report/{clean}')
|
||||
|
||||
return candidates
|
||||
|
||||
def get_template_sources(self, template_name):
|
||||
"""Yield Origin objects for templates found in default storage."""
|
||||
for candidate in self.get_storage_candidates(template_name):
|
||||
try:
|
||||
if default_storage.exists(candidate):
|
||||
yield Origin(
|
||||
name=candidate, template_name=template_name, loader=self
|
||||
)
|
||||
return
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
def get_contents(self, origin):
|
||||
"""Return template contents from default storage."""
|
||||
try:
|
||||
with default_storage.open(origin.name, 'r') as fp:
|
||||
content = fp.read()
|
||||
if isinstance(content, bytes):
|
||||
content = content.decode('utf-8')
|
||||
return content
|
||||
except Exception as exc:
|
||||
raise TemplateDoesNotExist(origin) from exc
|
||||
|
||||
|
||||
class InvenTreeTemplateLoader(CachedLoader):
|
||||
"""Custom template loader which bypasses cache for PDF export."""
|
||||
|
||||
@@ -20,6 +76,8 @@ class InvenTreeTemplateLoader(CachedLoader):
|
||||
skip_cache_dirs = [
|
||||
os.path.abspath(os.path.join(settings.MEDIA_ROOT, 'report')),
|
||||
os.path.abspath(os.path.join(settings.MEDIA_ROOT, 'label')),
|
||||
'report/',
|
||||
'label/',
|
||||
'snippets/',
|
||||
]
|
||||
|
||||
|
||||
@@ -102,13 +102,19 @@ class TemplateUploadMixin:
|
||||
@property
|
||||
def template_name(self):
|
||||
"""Return the filename of the template associated with this model class."""
|
||||
template = getattr(self, self.TEMPLATE_FIELD).name
|
||||
template_file = getattr(self, self.TEMPLATE_FIELD, None)
|
||||
if not template_file or not template_file.name:
|
||||
return ''
|
||||
|
||||
template = template_file.name
|
||||
template = template.replace('/', os.path.sep)
|
||||
template = template.replace('\\', os.path.sep)
|
||||
|
||||
template = settings.MEDIA_ROOT.joinpath(template)
|
||||
local_path = settings.MEDIA_ROOT.joinpath(template)
|
||||
if local_path.exists():
|
||||
return str(local_path)
|
||||
|
||||
return str(template)
|
||||
return template_file.name
|
||||
|
||||
@property
|
||||
def extension(self):
|
||||
@@ -251,6 +257,70 @@ class ReportTemplateBase(
|
||||
template_string = Template(self.filename_pattern)
|
||||
return template_string.render(Context(context))
|
||||
|
||||
def get_template_content(self) -> str:
|
||||
"""Load template file content directly from storage backend.
|
||||
|
||||
Supports cloud/remote storage (e.g. S3, SFTP) and resolves potential
|
||||
storage prefix mismatches.
|
||||
"""
|
||||
template_file = getattr(self, self.TEMPLATE_FIELD, None)
|
||||
if not template_file or not template_file.name:
|
||||
raise TemplateDoesNotExist(getattr(self, 'name', 'template'))
|
||||
|
||||
storage = getattr(template_file, 'storage', default_storage)
|
||||
name = template_file.name
|
||||
|
||||
# Candidates to resolve storage path prefix mismatches
|
||||
candidates = [name]
|
||||
norm_name = name.replace('\\', '/').lstrip('/')
|
||||
if norm_name not in candidates:
|
||||
candidates.append(norm_name)
|
||||
|
||||
location = getattr(storage, 'location', '') or ''
|
||||
location = str(location).replace('\\', '/').strip('/')
|
||||
if location:
|
||||
if norm_name.startswith(f'{location}/'):
|
||||
stripped = norm_name[len(location) + 1 :]
|
||||
if stripped not in candidates:
|
||||
candidates.append(stripped)
|
||||
else:
|
||||
prefixed = f'{location}/{norm_name}'
|
||||
if prefixed not in candidates:
|
||||
candidates.append(prefixed)
|
||||
|
||||
if hasattr(self, 'SUBDIR') and self.SUBDIR:
|
||||
subdir_prefix = f'report/{self.SUBDIR}/'
|
||||
if norm_name.startswith(subdir_prefix):
|
||||
short_name = norm_name[len('report/') :]
|
||||
if short_name not in candidates:
|
||||
candidates.append(short_name)
|
||||
|
||||
file_obj = None
|
||||
for candidate in candidates:
|
||||
try:
|
||||
if storage.exists(candidate):
|
||||
file_obj = storage.open(candidate, 'r')
|
||||
break
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if file_obj is None:
|
||||
try:
|
||||
file_obj = template_file.open('r')
|
||||
except Exception as exc:
|
||||
raise TemplateDoesNotExist(
|
||||
f"Template '{name}' not found in storage"
|
||||
) from exc
|
||||
|
||||
try:
|
||||
content = file_obj.read()
|
||||
if isinstance(content, bytes):
|
||||
content = content.decode('utf-8')
|
||||
return content
|
||||
finally:
|
||||
if hasattr(file_obj, 'close'):
|
||||
file_obj.close()
|
||||
|
||||
def render_as_string(
|
||||
self, instance: models.Model, context: Optional[dict] = None, **kwargs
|
||||
) -> str:
|
||||
@@ -266,7 +336,15 @@ class ReportTemplateBase(
|
||||
if context is None:
|
||||
context = self.get_context(instance, **kwargs)
|
||||
|
||||
return render_to_string(self.template_name, context)
|
||||
try:
|
||||
return render_to_string(self.template_name, context)
|
||||
except TemplateDoesNotExist:
|
||||
content = self.get_template_content()
|
||||
from django.template import engines
|
||||
|
||||
engine = engines['django']
|
||||
template = engine.from_string(content)
|
||||
return template.render(context)
|
||||
|
||||
def render(
|
||||
self, instance: models.Model, context: Optional[dict] = None, **kwargs
|
||||
|
||||
@@ -13,7 +13,8 @@ from django.contrib.contenttypes.models import ContentType
|
||||
from django.core.cache import cache
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.core.files.base import ContentFile
|
||||
from django.core.files.storage import default_storage
|
||||
from django.core.files.storage import Storage, default_storage
|
||||
from django.template import Context, TemplateDoesNotExist, engines
|
||||
from django.template.loader import render_to_string
|
||||
from django.test import TestCase
|
||||
from django.urls import reverse
|
||||
@@ -27,6 +28,7 @@ from build.models import Build
|
||||
from common.models import Attachment, Note
|
||||
from common.settings import set_global_setting
|
||||
from InvenTree.config import get_base_dir
|
||||
from InvenTree.template import InvenTreeStorageTemplateLoader
|
||||
from InvenTree.unit_test import AdminTestCase, InvenTreeAPITestCase
|
||||
from order.models import PurchaseOrder, ReturnOrder, SalesOrder
|
||||
from part.models import Part
|
||||
@@ -1543,3 +1545,151 @@ class DefaultTemplateFileTest(TestCase):
|
||||
# Size and content must match the raw bytes exactly
|
||||
self.assertEqual(content_file.size, len(raw))
|
||||
self.assertEqual(content_file.read(), raw)
|
||||
|
||||
|
||||
class MockS3Storage(Storage):
|
||||
"""Mock storage simulating S3 backend with prefix/location."""
|
||||
|
||||
def __init__(self, location='inventree-server'):
|
||||
"""Initialize mock storage with a location."""
|
||||
self.location = location
|
||||
self.files = {}
|
||||
|
||||
def _save(self, name, content):
|
||||
data = content.read() if hasattr(content, 'read') else content
|
||||
if isinstance(data, str):
|
||||
data = data.encode('utf-8')
|
||||
self.files[name] = data
|
||||
return name
|
||||
|
||||
def _open(self, name, mode='rb'):
|
||||
if name not in self.files:
|
||||
raise FileNotFoundError(f'{name} not found in MockS3Storage')
|
||||
return ContentFile(self.files[name], name=name)
|
||||
|
||||
def exists(self, name):
|
||||
"""Check if file exists in mock storage."""
|
||||
return name in self.files
|
||||
|
||||
def path(self, name):
|
||||
"""Raise NotImplementedError to simulate S3 remote storage."""
|
||||
raise NotImplementedError("This backend doesn't support absolute paths.")
|
||||
|
||||
|
||||
class S3ReportTemplateTest(TestCase):
|
||||
"""Unit tests for S3 cloud storage report template loading (Issue #12426)."""
|
||||
|
||||
def setUp(self):
|
||||
"""Setup test fixtures."""
|
||||
super().setUp()
|
||||
self.storage = MockS3Storage(location='inventree-server')
|
||||
|
||||
def test_template_name_remote_storage(self):
|
||||
"""Test template_name property when file only exists in remote cloud storage."""
|
||||
template = ReportTemplate(
|
||||
name='Remote S3 Report',
|
||||
description='Test S3 report loading',
|
||||
model_type='part',
|
||||
)
|
||||
template.template.name = 'report/report_template/remote_template.html'
|
||||
template.template.storage = self.storage
|
||||
|
||||
# Since the file does not exist locally under settings.MEDIA_ROOT,
|
||||
# template_name should return the relative storage path, not a non-existent MEDIA_ROOT path.
|
||||
self.assertEqual(
|
||||
template.template_name, 'report/report_template/remote_template.html'
|
||||
)
|
||||
|
||||
def test_get_template_content_with_prefix(self):
|
||||
"""Test get_template_content resolves templates with bucket/location prefix."""
|
||||
template = ReportTemplate(
|
||||
name='Prefixed S3 Report',
|
||||
description='Test prefix resolution',
|
||||
model_type='part',
|
||||
)
|
||||
template.template.name = 'report/report_template/test_prefix.html'
|
||||
template.template.storage = self.storage
|
||||
|
||||
# S3 storage stores file with location prefix
|
||||
content = b'<h1>Report: {{ part.name }}</h1>'
|
||||
self.storage.files[
|
||||
'inventree-server/report/report_template/test_prefix.html'
|
||||
] = content
|
||||
|
||||
result = template.get_template_content()
|
||||
self.assertEqual(result, '<h1>Report: {{ part.name }}</h1>')
|
||||
|
||||
def test_get_template_content_without_prefix(self):
|
||||
"""Test get_template_content when template file is stored without location prefix."""
|
||||
template = ReportTemplate(
|
||||
name='Direct S3 Report',
|
||||
description='Test direct resolution',
|
||||
model_type='part',
|
||||
)
|
||||
template.template.name = 'report/report_template/direct.html'
|
||||
template.template.storage = self.storage
|
||||
|
||||
content = b'<p>Direct template</p>'
|
||||
self.storage.files['report/report_template/direct.html'] = content
|
||||
|
||||
result = template.get_template_content()
|
||||
self.assertEqual(result, '<p>Direct template</p>')
|
||||
|
||||
def test_get_template_content_missing(self):
|
||||
"""Test get_template_content raises TemplateDoesNotExist when file not in storage."""
|
||||
template = ReportTemplate(
|
||||
name='Missing S3 Report',
|
||||
description='Test missing template',
|
||||
model_type='part',
|
||||
)
|
||||
template.template.name = 'report/report_template/missing.html'
|
||||
template.template.storage = self.storage
|
||||
|
||||
with self.assertRaises(TemplateDoesNotExist):
|
||||
template.get_template_content()
|
||||
|
||||
def test_render_as_string_fallback(self):
|
||||
"""Test render_as_string falls back to storage when TemplateDoesNotExist is caught."""
|
||||
template = ReportTemplate(
|
||||
name='Render S3 Report',
|
||||
description='Test render fallback',
|
||||
model_type='part',
|
||||
)
|
||||
template.template.name = 'report/report_template/render_test.html'
|
||||
template.template.storage = self.storage
|
||||
|
||||
# Put template into S3 storage with location prefix
|
||||
self.storage.files[
|
||||
'inventree-server/report/report_template/render_test.html'
|
||||
] = b'<div>Hello {{ test_val }}</div>'
|
||||
|
||||
rendered = template.render_as_string(None, context={'test_val': 'World'})
|
||||
self.assertEqual(rendered, '<div>Hello World</div>')
|
||||
|
||||
def test_storage_template_loader_resolution(self):
|
||||
"""Test InvenTreeStorageTemplateLoader loads templates from default_storage."""
|
||||
with patch('InvenTree.template.default_storage', self.storage):
|
||||
self.storage.files['inventree-server/report/snippets/header.html'] = (
|
||||
b'<header>{{ title }}</header>'
|
||||
)
|
||||
|
||||
loader = InvenTreeStorageTemplateLoader(engines['django'].engine)
|
||||
tmpl = loader.get_template('report/snippets/header.html')
|
||||
self.assertIsNotNone(tmpl)
|
||||
rendered = tmpl.render(Context({'title': 'S3 Header Test'}))
|
||||
self.assertEqual(rendered, '<header>S3 Header Test</header>')
|
||||
|
||||
with self.assertRaises(TemplateDoesNotExist):
|
||||
loader.get_template('does/not/exist.html')
|
||||
|
||||
def test_render_to_string_with_storage_loader(self):
|
||||
"""Test Django render_to_string resolves via InvenTreeStorageTemplateLoader."""
|
||||
with patch('InvenTree.template.default_storage', self.storage):
|
||||
self.storage.files['inventree-server/report/custom_snippet.html'] = (
|
||||
b'<span>User: {{ username }}</span>'
|
||||
)
|
||||
|
||||
result = render_to_string(
|
||||
'report/custom_snippet.html', {'username': 'testuser'}
|
||||
)
|
||||
self.assertEqual(result, '<span>User: testuser</span>')
|
||||
Reference in new issue
Block a user