fix(pdfa): address code review - validate pdfa_format, add S3 comment, add format test
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
@@ -59,6 +59,11 @@ def _convert_pdf_to_pdfa(input_path: str, output_path: str, pdfa_format: str = "
|
|||||||
Returns:
|
Returns:
|
||||||
True if conversion succeeded, False otherwise.
|
True if conversion succeeded, False otherwise.
|
||||||
"""
|
"""
|
||||||
|
# Validate format to prevent argument injection via output-type
|
||||||
|
if pdfa_format not in ("1", "2", "3"):
|
||||||
|
logger.error(f"[convert_to_pdfa] Invalid pdfa_format: {pdfa_format}")
|
||||||
|
return False
|
||||||
|
|
||||||
ocrmypdf_bin = shutil.which("ocrmypdf")
|
ocrmypdf_bin = shutil.which("ocrmypdf")
|
||||||
if not ocrmypdf_bin:
|
if not ocrmypdf_bin:
|
||||||
logger.error("[convert_to_pdfa] ocrmypdf binary not found on PATH")
|
logger.error("[convert_to_pdfa] ocrmypdf binary not found on PATH")
|
||||||
@@ -186,7 +191,8 @@ def _compute_pdfa_folder_overrides() -> dict[str, str]:
|
|||||||
base = getattr(settings, folder_attr, "") or ""
|
base = getattr(settings, folder_attr, "") or ""
|
||||||
overrides[provider] = f"{base.rstrip('/')}/{subfolder}" if base else subfolder
|
overrides[provider] = f"{base.rstrip('/')}/{subfolder}" if base else subfolder
|
||||||
|
|
||||||
# S3: append subfolder to prefix
|
# S3: append subfolder to prefix (trailing slash is required by S3 convention
|
||||||
|
# where "folder" paths are key prefixes, unlike path-based providers above)
|
||||||
s3_prefix = getattr(settings, "s3_folder_prefix", "") or ""
|
s3_prefix = getattr(settings, "s3_folder_prefix", "") or ""
|
||||||
overrides["s3"] = f"{s3_prefix.rstrip('/')}/{subfolder}/"
|
overrides["s3"] = f"{s3_prefix.rstrip('/')}/{subfolder}/"
|
||||||
|
|
||||||
|
|||||||
@@ -73,6 +73,11 @@ class TestConvertPdfToPdfa:
|
|||||||
cmd = mock_run.call_args[0][0]
|
cmd = mock_run.call_args[0][0]
|
||||||
assert f"pdfa-{fmt}" in cmd
|
assert f"pdfa-{fmt}" in cmd
|
||||||
|
|
||||||
|
def test_invalid_pdfa_format_rejected(self):
|
||||||
|
"""Test that invalid PDF/A format values are rejected."""
|
||||||
|
result = _convert_pdf_to_pdfa("/input.pdf", "/output.pdf", "invalid")
|
||||||
|
assert result is False
|
||||||
|
|
||||||
@patch("app.tasks.convert_to_pdfa.subprocess.run")
|
@patch("app.tasks.convert_to_pdfa.subprocess.run")
|
||||||
@patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/ocrmypdf")
|
@patch("app.tasks.convert_to_pdfa.shutil.which", return_value="/usr/bin/ocrmypdf")
|
||||||
def test_conversion_failure_empty_stderr(self, mock_which, mock_run):
|
def test_conversion_failure_empty_stderr(self, mock_which, mock_run):
|
||||||
|
|||||||
Reference in New Issue
Block a user