diff --git a/cloud_storage/__init__.py b/cloud_storage/__init__.py index eb6aa9e..a7d8fe7 100644 --- a/cloud_storage/__init__.py +++ b/cloud_storage/__init__.py @@ -4,8 +4,11 @@ __version__ = "15.7.4" +import json + import frappe import frappe.desk.form.load +from frappe.core.doctype.communication.communication import Communication from frappe.core.doctype.file import file as file_module from frappe.query_builder import DocType @@ -41,3 +44,57 @@ def patched_get_attachments(dt, dn): frappe.desk.form.load.get_attachments = patched_get_attachments + + +ATTACHMENT_FIELDS = ["name", "file_name", "file_url", "is_private"] + + +def get_files_for_document(dt, dn, fields=ATTACHMENT_FIELDS): + """All File rows that belong to a document, by `attached_to_*` or by File Association. + """ + files = frappe.get_all( + "File", fields=fields, filters={"attached_to_doctype": dt, "attached_to_name": str(dn)} + ) + if "cloud_storage" not in frappe.get_installed_apps(): + return files + + seen = {f.name for f in files} + linked = frappe.get_all( + "File Association", filters={"link_doctype": dt, "link_name": str(dn)}, pluck="parent" + ) + missing = [name for name in set(linked) if name not in seen] + if missing: + files += frappe.get_all("File", fields=fields, filters={"name": ["in", missing]}) + return files + + +def patched_get_communications(doctype, name, start=0, limit=20): + """ + HASH: 48366c6ecbad44ed24e6d02bdd8f8f189ce58927 + REPO: https://github.com/frappe/frappe + PATH: frappe/desk/form/load.py + METHOD: _get_communications + + Core looks attachments up by `attached_to_*` only, so when the same files are emailed + from one document several times only the last email lists them in the timeline. + """ + communications = frappe.desk.form.load.get_communication_data(doctype, name, start, limit) + for c in communications: + if c.communication_type in ("Communication", "Automated Message"): + c.attachments = json.dumps(get_files_for_document("Communication", c.name)) + + return communications + + +def patched_communication_get_attachments(self): + """ + HASH: 48366c6ecbad44ed24e6d02bdd8f8f189ce58927 + REPO: https://github.com/frappe/frappe + PATH: frappe/core/doctype/communication/communication.py + METHOD: get_attachments + """ + return get_files_for_document(self.DOCTYPE, self.name) + + +frappe.desk.form.load._get_communications = patched_get_communications +Communication.get_attachments = patched_communication_get_attachments diff --git a/cloud_storage/tests/test_communication_attachments.py b/cloud_storage/tests/test_communication_attachments.py new file mode 100644 index 0000000..ad07390 --- /dev/null +++ b/cloud_storage/tests/test_communication_attachments.py @@ -0,0 +1,88 @@ +# Copyright (c) 2026, AgriTheory and contributors +# For license information, please see license.txt + +import json +from io import BytesIO + +import frappe +import pytest + +from cloud_storage import get_files_for_document + +pytestmark = pytest.mark.usefixtures("mocked_s3_client") + + +def make_communication(subject): + return frappe.get_doc( + { + "doctype": "Communication", + "communication_type": "Communication", + "communication_medium": "Email", + "sent_or_received": "Sent", + "status": "Linked", + "subject": subject, + "sender": "Administrator", + "recipients": "supplier@example.com", + "content": subject, + "reference_doctype": "Role", + "reference_name": "System Manager", + } + ).insert(ignore_permissions=True) + + +def attach(communication, file_name, content): + # what frappe.core.doctype.communication.email.add_attachments does for a sent email + return frappe.get_doc( + { + "doctype": "File", + "file_name": file_name, + "attached_to_doctype": "Communication", + "attached_to_name": communication.name, + "folder": "Home/Attachments", + "content": content, + "decode": False, + "is_private": 1, + } + ).save(ignore_permissions=True) + + +def test_same_attachment_emailed_twice_is_listed_on_both_communications(): + """The second email's File is merged into the first one and steals `attached_to`, so core's + `attached_to`-based lookups list the attachment on one Communication only.""" + from PIL import Image + + frappe.set_user("Administrator") + buffer = BytesIO() + Image.new("RGB", (5, 9), (200, 30, 90)).save(buffer, format="PNG") + content = buffer.getvalue() + + first = make_communication("RFQ to supplier 1") + second = make_communication("RFQ to supplier 2") + try: + attach(first, "drawing.png", content) + attach(second, "drawing.png", content) + + # one physical File survives, now attached_to the second Communication + owners = frappe.get_all( + "File", filters={"file_name": "drawing.png"}, pluck="attached_to_name" + ) + assert owners == [second.name] + + for comm in (first, second): + files = get_files_for_document("Communication", comm.name) + assert [f.file_name for f in files] == ["drawing.png"] + assert [f.file_name for f in comm.get_attachments()] == ["drawing.png"] + + # what the form timeline renders for each email + timeline = { + c.name: json.loads(c.attachments) + for c in frappe.desk.form.load._get_communications("Role", "System Manager", limit=50) + if c.name in (first.name, second.name) + } + assert len(timeline) == 2 + for files in timeline.values(): + assert len(files) == 1 + assert "?key=" in files[0]["file_url"] + finally: + for comm in (first, second): + frappe.delete_doc("Communication", comm.name, ignore_permissions=True, force=True)