diff --git a/press/api/email.py b/press/api/email.py
index 41e03d71dbb..7ec007073e3 100644
--- a/press/api/email.py
+++ b/press/api/email.py
@@ -79,16 +79,13 @@ def setup(site):
@frappe.whitelist(allow_guest=True)
-def get_analytics(**data):
+def get_analytics(month: int, status: str, site: str, key: str):
"""
send data for a specific month
"""
- month = data.get("month")
year = datetime.now().year
last_day = calendar.monthrange(year, int(month))[1]
- status = data.get("status")
- site = data.get("site")
- subscription_key = data.get("key")
+ subscription_key = key
for value in (site, subscription_key):
if not value or not isinstance(value, str):
@@ -207,25 +204,25 @@ def check_recipients(recipients: str | list[str]):
@frappe.whitelist(allow_guest=True)
-def send_mime_mail(**data):
+def send_mime_mail(data: str):
"""
send api request to mailgun
"""
files = frappe._dict(frappe.request.files)
- data = json.loads(data["data"])
+ data_dict = json.loads(data)
- validate_plan(data["sk_mail"])
+ validate_plan(data_dict["sk_mail"])
api_key, domain = frappe.db.get_value("Press Settings", None, ["mailgun_api_key", "root_domain"])
message: bytes = files["mime"].read()
check_spam(message)
- check_recipients(data["recipients"])
+ check_recipients(data_dict["recipients"])
resp = requests.post(
f"https://api.mailgun.net/v3/{domain}/messages.mime",
auth=("api", f"{api_key}"),
- data={"to": data["recipients"], "v:sk_mail": data["sk_mail"]},
+ data={"to": data_dict["recipients"], "v:sk_mail": data_dict["sk_mail"]},
files={"message": message},
)
@@ -234,7 +231,7 @@ def send_mime_mail(**data):
if resp.status_code == 400:
err_msg: str = resp.json().get("message", "Invalid request")
frappe.throw(f"Something went wrong with sending emails: {err_msg}", InvalidEmail)
- log_error("Email Delivery Service: Sending error", response=resp.text, data=data, message=message)
+ log_error("Email Delivery Service: Sending error", response=resp.text, data=data_dict, message=message)
frappe.throw(
"Something went wrong with sending emails. Please try again later or raise a support ticket with support.frappe.io",
EmailSendError,
diff --git a/press/api/github.py b/press/api/github.py
index ab70ea4c3cd..89fa663a74e 100644
--- a/press/api/github.py
+++ b/press/api/github.py
@@ -302,7 +302,7 @@ def repositories(installation, token):
@frappe.whitelist()
-def repository(owner, name, installation=None):
+def repository(owner: str, name: str, installation: str | None = None):
token = ""
if not installation:
token = frappe.db.get_value("Press Settings", "github_access_token")
@@ -337,7 +337,7 @@ def repository(owner, name, installation=None):
@frappe.whitelist()
-def app(owner, repository, branch, installation=None):
+def app(owner: str, repository: str, branch: str, installation: str | None = None):
headers = get_auth_headers(installation)
response = requests.get(
f"https://api.github.com/repos/{owner}/{repository}/branches/{branch}",
@@ -381,7 +381,7 @@ def app(owner, repository, branch, installation=None):
@frappe.whitelist()
-def branches(owner, name, installation=None, app_source=None):
+def branches(owner: str, name: str, installation: str | None = None, app_source: str | None = None):
"""
Return ALL branches for the repo, following GitHub pagination.
"""
@@ -390,7 +390,7 @@ def branches(owner, name, installation=None, app_source=None):
headers = get_auth_headers(installation)
- out = []
+ out: list[dict] = []
page = 1
while True:
resp = requests.get(
@@ -487,7 +487,7 @@ def _get_app_name_and_title_from_hooks(
branch_info,
headers,
tree,
-) -> tuple[str, str] | None:
+) -> tuple[str, str]:
reason_for_invalidation = f"Files {frappe.bold('hooks.py or patches.txt')} not found."
for directory, files in tree.items():
if not files:
@@ -524,7 +524,7 @@ def _get_app_name_and_title_from_hooks(
break
frappe.throw(f"Not a valid Frappe App! {reason_for_invalidation}")
- return None
+ raise # for mypy: NoReturn
def _generate_files_tree(files):
diff --git a/press/api/monitoring.py b/press/api/monitoring.py
index 131f7c8d18b..8faef742b18 100644
--- a/press/api/monitoring.py
+++ b/press/api/monitoring.py
@@ -159,7 +159,7 @@ def get_targets_method_rate_limit() -> int:
@frappe.whitelist(allow_guest=True)
@rate_limit(limit=get_targets_method_rate_limit, seconds=MONITORING_ENDPOINT_RATE_LIMIT_WINDOW_SECONDS)
-def targets(token=None):
+def targets(token: str | None = None):
if not token:
frappe.throw_permission_error()
monitor_token = frappe.db.get_single_value("Press Settings", "monitor_token", cache=True)
diff --git a/press/api/tests/test_email.py b/press/api/tests/test_email.py
new file mode 100644
index 00000000000..631d8c6d73f
--- /dev/null
+++ b/press/api/tests/test_email.py
@@ -0,0 +1,124 @@
+# Copyright (c) 2024, Frappe and contributors
+# For license information, please see license.txt
+
+from __future__ import annotations
+
+import io
+import json
+from unittest.mock import Mock, patch
+
+import frappe
+from frappe.tests.test_api import FrappeAPITestCase, make_request
+
+
+class TestSendMimeMail(FrappeAPITestCase):
+ """Test send_mime_mail endpoint with the same parameters email_delivery_service uses."""
+
+ ENDPOINT = "/api/method/press.api.email.send_mime_mail"
+
+ def _post_mime_mail(
+ self,
+ data: dict,
+ mime_content: bytes = b"MIME-Version: 1.0\r\nContent-Type: text/plain\r\n\r\nTest email body",
+ ):
+ """Send a multipart form request with a MIME file, matching email_delivery_service.
+
+ The email_delivery_service app sends:
+ requests.post(url, data={"data": json.dumps(data)}, files={"mime": msg})
+ """
+ return make_request(
+ target=self.TEST_CLIENT.post,
+ args=(self.ENDPOINT,),
+ kwargs={
+ "data": {
+ "data": json.dumps(data),
+ "mime": (io.BytesIO(mime_content), "message.eml"),
+ },
+ "content_type": "multipart/form-data",
+ "buffered": True,
+ },
+ )
+
+ @patch("press.api.email.validate_plan")
+ @patch("press.api.email.check_spam")
+ @patch("press.api.email.requests.post")
+ def test_send_mime_mail_success(self, mock_mailgun_post, mock_check_spam, mock_validate_plan):
+ """Test that send_mime_mail correctly parses the data parameter as a JSON string,
+ using the same request format as email_delivery_service."""
+ mock_mailgun_response = Mock()
+ mock_mailgun_response.status_code = 200
+ mock_mailgun_post.return_value = mock_mailgun_response
+
+ frappe.db.set_single_value("Press Settings", "mailgun_api_key", "test-key")
+ frappe.db.set_single_value("Press Settings", "root_domain", "example.com")
+
+ data = {
+ "sender": "sender@example.com",
+ "recipients": "recipient@example.com",
+ "sk_mail": "test-secret-key",
+ "site": "test.frappe.cloud",
+ }
+ mime_content = b"MIME-Version: 1.0\r\nContent-Type: text/plain\r\n\r\nTest email body"
+
+ response = self._post_mime_mail(data, mime_content)
+
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.json["message"], "Sending")
+
+ mock_validate_plan.assert_called_once_with("test-secret-key")
+ mock_check_spam.assert_called_once_with(mime_content)
+ mock_mailgun_post.assert_called_once()
+
+ call_kwargs = mock_mailgun_post.call_args
+ self.assertEqual(call_kwargs.kwargs["data"]["to"], "recipient@example.com")
+ self.assertEqual(call_kwargs.kwargs["data"]["v:sk_mail"], "test-secret-key")
+
+ @patch("press.api.email.validate_plan")
+ @patch("press.api.email.check_spam", new=Mock())
+ @patch("press.api.email.requests.post")
+ def test_send_mime_mail_parses_json_string_data(self, mock_mailgun_post, mock_validate_plan):
+ """Verify the data parameter is correctly parsed as a JSON string (not a dict)."""
+ mock_mailgun_response = Mock()
+ mock_mailgun_response.status_code = 200
+ mock_mailgun_post.return_value = mock_mailgun_response
+
+ frappe.db.set_single_value("Press Settings", "mailgun_api_key", "test-key")
+ frappe.db.set_single_value("Press Settings", "root_domain", "example.com")
+
+ data = {
+ "sender": "test@site.com",
+ "recipients": "user@domain.com",
+ "sk_mail": "sk-12345",
+ "site": "mysite.frappe.cloud",
+ }
+
+ response = self._post_mime_mail(data)
+
+ self.assertEqual(response.status_code, 200)
+ # Verify validate_plan received the correct sk_mail from the parsed JSON
+ mock_validate_plan.assert_called_once_with("sk-12345")
+
+ @patch("press.api.email.validate_plan", new=Mock())
+ @patch("press.api.email.check_spam", new=Mock())
+ def test_send_mime_mail_mailgun_400_error(self):
+ """Test that a 400 response from mailgun raises InvalidEmail."""
+
+ frappe.db.set_single_value("Press Settings", "mailgun_api_key", "test-key")
+ frappe.db.set_single_value("Press Settings", "root_domain", "example.com")
+
+ data = {
+ "sender": "sender@example.com",
+ "recipients": "invalid",
+ "sk_mail": "test-secret-key",
+ "site": "test.frappe.cloud",
+ }
+
+ response = self._post_mime_mail(data)
+
+ self.assertIn(response.status_code, (400, 417))
+
+ def test_email_ping(self):
+ """Test the email_ping endpoint."""
+ response = self.get(self.method("press.api.email.email_ping"))
+ self.assertEqual(response.status_code, 200)
+ self.assertEqual(response.json["message"], "pong")
diff --git a/press/press/doctype/site/site.py b/press/press/doctype/site/site.py
index a3226ab2525..1106f7bdade 100644
--- a/press/press/doctype/site/site.py
+++ b/press/press/doctype/site/site.py
@@ -1309,7 +1309,7 @@ def ready_for_move(self):
def check_fatal_site_update(self):
if self.fatal_site_update:
frappe.throw(
- "Site has encountered a fatal error during last update. Please open a ticket on our support portal with the error details to resolve the issue.",
+ "Site has encountered a fatal error during last update. Please open a ticket on our support portal with the error details to resolve the issue.",
)
@dashboard_whitelist()
diff --git a/press/press/doctype/site_update/site_update.py b/press/press/doctype/site_update/site_update.py
index 819b943b856..1ec9a6b4060 100644
--- a/press/press/doctype/site_update/site_update.py
+++ b/press/press/doctype/site_update/site_update.py
@@ -237,7 +237,7 @@ def validate_apps(self):
def before_insert(self):
self.backup_type = "Logical"
- site: "Site" = frappe.get_cached_doc("Site", self.site)
+ site: "Site" = frappe.get_doc("Site", self.site)
site.check_move_scheduled()
site.check_fatal_site_update()
diff --git a/press/utils/dns.py b/press/utils/dns.py
index 895eed378a8..650aec4034d 100644
--- a/press/utils/dns.py
+++ b/press/utils/dns.py
@@ -145,7 +145,7 @@ def accessible_link_substr(provider: str):
except requests.RequestException:
return None
else:
- return f'{provider}'
+ return f'{provider}'
@redis_cache()