Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 57 additions & 0 deletions cloud_storage/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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
88 changes: 88 additions & 0 deletions cloud_storage/tests/test_communication_attachments.py
Original file line number Diff line number Diff line change
@@ -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)
Loading