Merge pull request #604 from christianlouis/copilot/configure-attachment-ingestion

feat(imap): fine-grained attachment ingestion profiles with per-category selection
2026-03-12 08:53:45 +01:00
parent 338f4fa9bd 2e087731b9
commit 7f644a2388
18 changed files with 1614 additions and 30 deletions
@@ -0,0 +1,159 @@
+"""Tests for app/api/imap_profiles.py and app/utils/allowed_types category helpers."""
+
+
+import pytest
+
+from app.utils.allowed_types import (
+    ALL_CATEGORIES,
+    DEFAULT_CATEGORIES,
+    FILE_TYPE_CATEGORIES,
+    get_allowed_types_for_categories,
+)
+
+
+@pytest.mark.unit
+class TestFileTypeCategories:
+    """Tests for FILE_TYPE_CATEGORIES and get_allowed_types_for_categories."""
+
+    def test_all_category_keys_present(self):
+        """Test that the six expected categories exist."""
+        assert set(FILE_TYPE_CATEGORIES.keys()) == {"pdf", "office", "opendocument", "text", "web", "images"}
+
+    def test_each_category_has_required_fields(self):
+        """Test that every category entry has label, description, mime_types, extensions."""
+        for key, info in FILE_TYPE_CATEGORIES.items():
+            assert "label" in info, f"Category '{key}' missing 'label'"
+            assert "description" in info, f"Category '{key}' missing 'description'"
+            assert "mime_types" in info, f"Category '{key}' missing 'mime_types'"
+            assert "extensions" in info, f"Category '{key}' missing 'extensions'"
+
+    def test_pdf_category_contains_pdf_mime(self):
+        """Test that the pdf category includes application/pdf."""
+        assert "application/pdf" in FILE_TYPE_CATEGORIES["pdf"]["mime_types"]
+        assert ".pdf" in FILE_TYPE_CATEGORIES["pdf"]["extensions"]
+
+    def test_images_category_contains_jpeg(self):
+        """Test that the images category includes image/jpeg."""
+        assert "image/jpeg" in FILE_TYPE_CATEGORIES["images"]["mime_types"]
+        assert ".jpg" in FILE_TYPE_CATEGORIES["images"]["extensions"]
+        assert ".png" in FILE_TYPE_CATEGORIES["images"]["extensions"]
+
+    def test_get_allowed_types_for_default_categories(self):
+        """Test that DEFAULT_CATEGORIES excludes image MIME types."""
+        mime_types, extensions = get_allowed_types_for_categories(DEFAULT_CATEGORIES)
+        assert "application/pdf" in mime_types
+        assert "application/msword" in mime_types
+        # images should NOT be in the default
+        assert "image/jpeg" not in mime_types
+        assert ".jpg" not in extensions
+
+    def test_get_allowed_types_for_all_categories(self):
+        """Test that ALL_CATEGORIES includes image MIME types."""
+        mime_types, extensions = get_allowed_types_for_categories(ALL_CATEGORIES)
+        assert "image/jpeg" in mime_types
+        assert ".jpg" in extensions
+        assert "application/pdf" in mime_types
+
+    def test_get_allowed_types_returns_frozensets(self):
+        """Test that returned sets are frozensets."""
+        mime_types, extensions = get_allowed_types_for_categories(["pdf"])
+        assert isinstance(mime_types, frozenset)
+        assert isinstance(extensions, frozenset)
+
+    def test_get_allowed_types_unknown_category_ignored(self):
+        """Test that unknown category keys are silently ignored."""
+        mime_types, extensions = get_allowed_types_for_categories(["pdf", "nonexistent_category"])
+        assert "application/pdf" in mime_types  # 'pdf' still works
+        # No crash for unknown key
+
+    def test_get_allowed_types_empty_list(self):
+        """Test empty category list returns empty sets."""
+        mime_types, extensions = get_allowed_types_for_categories([])
+        assert mime_types == frozenset()
+        assert extensions == frozenset()
+
+    def test_default_categories_excludes_images(self):
+        """Test that DEFAULT_CATEGORIES does not include 'images'."""
+        assert "images" not in DEFAULT_CATEGORIES
+
+    def test_all_categories_includes_images(self):
+        """Test that ALL_CATEGORIES includes 'images'."""
+        assert "images" in ALL_CATEGORIES
+
+    def test_all_categories_is_superset_of_default(self):
+        """Test that ALL_CATEGORIES contains all DEFAULT_CATEGORIES."""
+        for cat in DEFAULT_CATEGORIES:
+            assert cat in ALL_CATEGORIES
+
+
+@pytest.mark.unit
+class TestImapProfilesApiLogic:
+    """Tests for ingestion profile validation helpers."""
+
+    def test_validate_categories_accepts_valid_keys(self):
+        """Test that valid category keys pass validation."""
+        from app.api.imap_profiles import _validate_categories
+
+        result = _validate_categories(["pdf", "office", "images"])
+        assert set(result) == {"pdf", "office", "images"}
+
+    def test_validate_categories_rejects_unknown_key(self):
+        """Test that unknown category keys raise 422."""
+        from fastapi import HTTPException
+
+        from app.api.imap_profiles import _validate_categories
+
+        with pytest.raises(HTTPException) as exc_info:
+            _validate_categories(["pdf", "nonexistent"])
+        assert exc_info.value.status_code == 422
+        assert "nonexistent" in str(exc_info.value.detail)
+
+    def test_validate_categories_deduplicates(self):
+        """Test that duplicate category keys are de-duplicated while preserving order."""
+        from app.api.imap_profiles import _validate_categories
+
+        result = _validate_categories(["pdf", "pdf", "office", "pdf"])
+        assert result == ["pdf", "office"]
+
+    def test_to_response_serializes_profile(self, tmp_path):
+        """Test _to_response produces expected dict shape."""
+        from unittest.mock import MagicMock
+
+        from app.api.imap_profiles import _to_response
+
+        profile = MagicMock()
+        profile.id = 42
+        profile.name = "My Profile"
+        profile.description = "Test description"
+        profile.owner_id = "user@example.com"
+        profile.allowed_categories = '["pdf","office"]'
+        profile.is_builtin = False
+        profile.created_at = None
+        profile.updated_at = None
+
+        result = _to_response(profile)
+        assert result["id"] == 42
+        assert result["name"] == "My Profile"
+        assert result["allowed_categories"] == ["pdf", "office"]
+        assert len(result["categories_detail"]) == 2
+        assert result["categories_detail"][0]["key"] == "pdf"
+        assert result["is_builtin"] is False
+
+    def test_to_response_handles_invalid_categories_json(self):
+        """Test _to_response gracefully handles invalid JSON in allowed_categories."""
+        from unittest.mock import MagicMock
+
+        from app.api.imap_profiles import _to_response
+
+        profile = MagicMock()
+        profile.id = 1
+        profile.name = "Broken"
+        profile.description = None
+        profile.owner_id = None
+        profile.allowed_categories = "this is not valid json {"
+        profile.is_builtin = True
+        profile.created_at = None
+        profile.updated_at = None
+
+        result = _to_response(profile)
+        assert result["allowed_categories"] == []
@@ -24,6 +24,7 @@ from app.tasks.imap_tasks import (
    release_lock,
    save_processed_emails,
 )
+from app.utils.allowed_types import ALL_CATEGORIES, DEFAULT_CATEGORIES, get_allowed_types_for_categories

 _TEST_CREDENTIAL = "pass"  # noqa: S105

@@ -179,6 +180,81 @@ class TestFetchAttachmentsAndEnqueue:
        assert result is True
        mock_convert.delay.assert_called_once()

+    @patch("app.tasks.imap_tasks.process_document")
+    @patch("app.tasks.imap_tasks.convert_to_pdf")
+    def test_skips_image_when_documents_only(self, mock_convert, mock_process):
+        """Test that image attachments are skipped with the default (documents-only) categories."""
+        msg = EmailMessage()
+        msg["Subject"] = "Photo"
+        msg.add_attachment(b"\xff\xd8\xff", maintype="image", subtype="jpeg", filename="photo.jpg")
+
+        doc_mime, doc_ext = get_allowed_types_for_categories(DEFAULT_CATEGORIES)
+        result = fetch_attachments_and_enqueue(msg, effective_mime_types=doc_mime, effective_extensions=doc_ext)
+        assert result is False
+        mock_process.delay.assert_not_called()
+        mock_convert.delay.assert_not_called()
+
+    @patch("app.tasks.imap_tasks.process_document")
+    @patch("app.tasks.imap_tasks.convert_to_pdf")
+    def test_processes_image_when_all_categories(self, mock_convert, mock_process, tmp_path):
+        """Test that image attachments are processed when images category is included."""
+        msg = EmailMessage()
+        msg["Subject"] = "Photo"
+        msg.add_attachment(b"\xff\xd8\xff", maintype="image", subtype="jpeg", filename="photo.jpg")
+
+        all_mime, all_ext = get_allowed_types_for_categories(ALL_CATEGORIES)
+        with patch("app.tasks.imap_tasks.settings") as mock_settings:
+            mock_settings.workdir = str(tmp_path)
+            result = fetch_attachments_and_enqueue(msg, effective_mime_types=all_mime, effective_extensions=all_ext)
+
+        assert result is True
+        mock_convert.delay.assert_called_once()
+
+    @patch("app.tasks.imap_tasks.process_document")
+    @patch("app.tasks.imap_tasks.convert_to_pdf")
+    def test_skips_image_with_image_extension_documents_only(self, mock_convert, mock_process):
+        """Test image files identified by extension are skipped with documents-only categories."""
+        msg = EmailMessage()
+        msg["Subject"] = "Screenshot"
+        msg.add_attachment(b"\x89PNG", maintype="application", subtype="octet-stream", filename="screenshot.png")
+
+        doc_mime, doc_ext = get_allowed_types_for_categories(DEFAULT_CATEGORIES)
+        result = fetch_attachments_and_enqueue(msg, effective_mime_types=doc_mime, effective_extensions=doc_ext)
+        assert result is False
+        mock_process.delay.assert_not_called()
+        mock_convert.delay.assert_not_called()
+
+    @patch("app.tasks.imap_tasks.process_document")
+    @patch("app.tasks.imap_tasks.convert_to_pdf")
+    def test_processes_pdf_regardless_of_categories(self, mock_convert, mock_process, tmp_path):
+        """Test that PDFs are always processed (pdf category always included in DEFAULT_CATEGORIES)."""
+        msg = EmailMessage()
+        msg["Subject"] = "Invoice"
+        msg.add_attachment(b"%PDF-1.4", maintype="application", subtype="pdf", filename="invoice.pdf")
+
+        doc_mime, doc_ext = get_allowed_types_for_categories(DEFAULT_CATEGORIES)
+        with patch("app.tasks.imap_tasks.settings") as mock_settings:
+            mock_settings.workdir = str(tmp_path)
+            result = fetch_attachments_and_enqueue(msg, effective_mime_types=doc_mime, effective_extensions=doc_ext)
+
+        assert result is True
+        mock_process.delay.assert_called_once()
+
+    @patch("app.tasks.imap_tasks.process_document")
+    @patch("app.tasks.imap_tasks.convert_to_pdf")
+    def test_uses_default_categories_when_no_types_provided(self, mock_convert, mock_process):
+        """Test that images are skipped when no effective_mime_types / extensions are passed (defaults to DEFAULT_CATEGORIES)."""
+        msg = EmailMessage()
+        msg["Subject"] = "Photo"
+        msg.add_attachment(b"\xff\xd8\xff", maintype="image", subtype="jpeg", filename="photo.jpg")
+
+        # No effective_mime_types passed → function defaults to DEFAULT_CATEGORIES (no images)
+        result = fetch_attachments_and_enqueue(msg)
+
+        assert result is False
+        mock_process.delay.assert_not_called()
+        mock_convert.delay.assert_not_called()
+

@pytest.mark.unit
 class TestEmailAlreadyHasLabel: