Skip to content

Commit 9b90568

Browse files
Venkatakrishnan774Venkatakrishnan774
committed
added test
1 parent 59426a3 commit 9b90568

1 file changed

Lines changed: 384 additions & 0 deletions

File tree

‎tests/outputmanagement/unit/test_email_client_mcp.py‎

Lines changed: 384 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,384 @@
1+
# SPDX-FileCopyrightText: 2024 SAP SE or an SAP affiliate company
2+
# SPDX-License-Identifier: Apache-2.0
3+
4+
"""Tests for EmailClient MCP integration."""
5+
6+
import pytest
7+
import asyncio
8+
from unittest.mock import Mock, AsyncMock, patch
9+
from sap_cloud_sdk.outputmanagement.clients.email_client import EmailClient
10+
11+
12+
class TestEmailClientMCP:
13+
"""Test EmailClient MCP integration methods."""
14+
15+
@pytest.fixture
16+
def email_client(self):
17+
"""Create an EmailClient instance for testing."""
18+
return EmailClient()
19+
20+
@pytest.fixture
21+
def sample_business_document(self):
22+
"""Sample business document for testing."""
23+
return {
24+
"PurchaseOrder": {
25+
"orderId": "PO-12345",
26+
"vendor": "ACME Corp",
27+
"total": 1500.00,
28+
"items": [
29+
{"product": "Widget A", "quantity": 10, "price": 100.00},
30+
{"product": "Widget B", "quantity": 5, "price": 100.00}
31+
]
32+
}
33+
}
34+
35+
@pytest.fixture
36+
def mock_mcp_tool(self):
37+
"""Create a mock MCP tool."""
38+
tool = Mock()
39+
tool.ainvoke = AsyncMock(return_value={"status": "success", "requestId": "req-123"})
40+
return tool
41+
42+
@pytest.mark.asyncio
43+
async def test_send_email_with_mcp_basic(self, email_client, sample_business_document, mock_mcp_tool):
44+
"""Test basic MCP email sending."""
45+
result = await email_client.send_email_with_mcp(
46+
tool_name="send_output_request",
47+
notification_template_key="PO_APPROVAL_NOTIFICATION",
48+
to_emails=["finance@company.com"],
49+
business_document=sample_business_document,
50+
mcp_tool=mock_mcp_tool
51+
)
52+
53+
assert result is not None
54+
assert result["status"] == "success"
55+
assert result["requestId"] == "req-123"
56+
mock_mcp_tool.ainvoke.assert_called_once()
57+
58+
@pytest.mark.asyncio
59+
async def test_send_email_with_mcp_with_cc(self, email_client, sample_business_document, mock_mcp_tool):
60+
"""Test MCP email sending with CC."""
61+
result = await email_client.send_email_with_mcp(
62+
tool_name="send_output_request",
63+
notification_template_key="PO_APPROVAL_NOTIFICATION",
64+
to_emails=["finance@company.com"],
65+
business_document=sample_business_document,
66+
cc_email="manager@company.com",
67+
mcp_tool=mock_mcp_tool
68+
)
69+
70+
assert result is not None
71+
mock_mcp_tool.ainvoke.assert_called_once()
72+
73+
# Verify CC was included in the payload
74+
call_args = mock_mcp_tool.ainvoke.call_args[0][0]
75+
assert "body" in call_args
76+
assert call_args["body"]["data"]["outputManagement"]["emailConfiguration"]["cc"] == ["manager@company.com"]
77+
78+
@pytest.mark.asyncio
79+
async def test_send_email_with_mcp_with_attachments(self, email_client, sample_business_document, mock_mcp_tool):
80+
"""Test MCP email sending with attachments."""
81+
attachment_urls = [
82+
"https://dms.example.com/browser/root?objectId=12345&cmisselector=content",
83+
"https://dms.example.com/browser/root?objectId=67890&cmisselector=content"
84+
]
85+
86+
result = await email_client.send_email_with_mcp(
87+
tool_name="send_output_request",
88+
notification_template_key="PO_APPROVAL_NOTIFICATION",
89+
to_emails=["finance@company.com"],
90+
business_document=sample_business_document,
91+
attachment_urls=attachment_urls,
92+
mcp_tool=mock_mcp_tool
93+
)
94+
95+
assert result is not None
96+
mock_mcp_tool.ainvoke.assert_called_once()
97+
98+
# Verify attachments were included in the payload
99+
call_args = mock_mcp_tool.ainvoke.call_args[0][0]
100+
assert "body" in call_args
101+
email_config = call_args["body"]["data"]["outputManagement"]["emailConfiguration"]
102+
assert "attachment" in email_config
103+
assert len(email_config["attachment"]["preGeneratedAttachments"]) == 2
104+
105+
@pytest.mark.asyncio
106+
async def test_send_email_with_mcp_traceparent_generated(self, email_client, sample_business_document, mock_mcp_tool):
107+
"""Test that traceparent is generated correctly."""
108+
result = await email_client.send_email_with_mcp(
109+
tool_name="send_output_request",
110+
notification_template_key="PO_APPROVAL_NOTIFICATION",
111+
to_emails=["finance@company.com"],
112+
business_document=sample_business_document,
113+
mcp_tool=mock_mcp_tool
114+
)
115+
116+
assert result is not None
117+
mock_mcp_tool.ainvoke.assert_called_once()
118+
119+
# Verify traceparent format (W3C Trace Context)
120+
call_args = mock_mcp_tool.ainvoke.call_args[0][0]
121+
assert "traceparent" in call_args
122+
traceparent = call_args["traceparent"]
123+
124+
# Format: 00-{trace_id}-{parent_id}-01
125+
parts = traceparent.split("-")
126+
assert len(parts) == 4
127+
assert parts[0] == "00" # version
128+
assert len(parts[1]) == 32 # trace_id (32 hex chars)
129+
assert len(parts[2]) == 16 # parent_id (16 hex chars)
130+
assert parts[3] == "01" # trace-flags
131+
132+
@pytest.mark.asyncio
133+
async def test_send_email_with_mcp_sender_subaccount_from_param(self, email_client, sample_business_document, mock_mcp_tool):
134+
"""Test sender_provider_subaccount_id from parameter."""
135+
result = await email_client.send_email_with_mcp(
136+
tool_name="send_output_request",
137+
notification_template_key="PO_APPROVAL_NOTIFICATION",
138+
to_emails=["finance@company.com"],
139+
business_document=sample_business_document,
140+
mcp_tool=mock_mcp_tool,
141+
sender_provider_subaccount_id="test-subaccount-123"
142+
)
143+
144+
assert result is not None
145+
mock_mcp_tool.ainvoke.assert_called_once()
146+
147+
# Verify sender_provider_subaccount_id was included
148+
call_args = mock_mcp_tool.ainvoke.call_args[0][0]
149+
assert "sender_provider_subaccount_id" in call_args
150+
assert call_args["sender_provider_subaccount_id"] == "test-subaccount-123"
151+
152+
@pytest.mark.asyncio
153+
@patch.dict('os.environ', {'APPFND_CONHOS_SUBACCOUNTID': 'env-subaccount-456'})
154+
async def test_send_email_with_mcp_sender_subaccount_from_env(self, email_client, sample_business_document, mock_mcp_tool):
155+
"""Test sender_provider_subaccount_id from environment variable."""
156+
result = await email_client.send_email_with_mcp(
157+
tool_name="send_output_request",
158+
notification_template_key="PO_APPROVAL_NOTIFICATION",
159+
to_emails=["finance@company.com"],
160+
business_document=sample_business_document,
161+
mcp_tool=mock_mcp_tool
162+
)
163+
164+
assert result is not None
165+
mock_mcp_tool.ainvoke.assert_called_once()
166+
167+
# Verify sender_provider_subaccount_id from env was used
168+
call_args = mock_mcp_tool.ainvoke.call_args[0][0]
169+
assert "sender_provider_subaccount_id" in call_args
170+
assert call_args["sender_provider_subaccount_id"] == "env-subaccount-456"
171+
172+
@pytest.mark.asyncio
173+
async def test_send_email_with_mcp_missing_tool_raises_error(self, email_client, sample_business_document):
174+
"""Test that missing MCP tool raises ValueError."""
175+
with pytest.raises(ValueError, match="mcp_tool parameter is required"):
176+
await email_client.send_email_with_mcp(
177+
tool_name="send_output_request",
178+
notification_template_key="PO_APPROVAL_NOTIFICATION",
179+
to_emails=["finance@company.com"],
180+
business_document=sample_business_document,
181+
mcp_tool=None
182+
)
183+
184+
@pytest.mark.asyncio
185+
async def test_send_email_with_mcp_tool_failure(self, email_client, sample_business_document):
186+
"""Test handling of MCP tool invocation failure."""
187+
mock_tool = Mock()
188+
mock_tool.ainvoke = AsyncMock(side_effect=Exception("MCP tool error"))
189+
190+
with pytest.raises(Exception, match="MCP tool invocation failed"):
191+
await email_client.send_email_with_mcp(
192+
tool_name="send_output_request",
193+
notification_template_key="PO_APPROVAL_NOTIFICATION",
194+
to_emails=["finance@company.com"],
195+
business_document=sample_business_document,
196+
mcp_tool=mock_tool
197+
)
198+
199+
@pytest.mark.asyncio
200+
async def test_send_email_with_mcp_payload_structure(self, email_client, sample_business_document, mock_mcp_tool):
201+
"""Test that the MCP payload has correct structure."""
202+
result = await email_client.send_email_with_mcp(
203+
tool_name="send_output_request",
204+
notification_template_key="PO_APPROVAL_NOTIFICATION",
205+
to_emails=["finance@company.com", "accounting@company.com"],
206+
business_document=sample_business_document,
207+
cc_email="manager@company.com",
208+
mcp_tool=mock_mcp_tool,
209+
sender_provider_subaccount_id="test-subaccount"
210+
)
211+
212+
assert result is not None
213+
mock_mcp_tool.ainvoke.assert_called_once()
214+
215+
# Verify complete payload structure
216+
call_args = mock_mcp_tool.ainvoke.call_args[0][0]
217+
218+
# Top-level keys
219+
assert "body" in call_args
220+
assert "traceparent" in call_args
221+
assert "sender_provider_subaccount_id" in call_args
222+
223+
# Body structure (CloudEvents format)
224+
body = call_args["body"]
225+
assert "source" in body
226+
assert "type" in body
227+
assert "data" in body
228+
229+
# Data structure
230+
data = body["data"]
231+
assert "outputManagement" in data
232+
assert "businessDocument" in data
233+
234+
# Output management structure
235+
output_mgmt = data["outputManagement"]
236+
assert "businessDocumentType" in output_mgmt
237+
assert "businessDocumentId" in output_mgmt
238+
assert "channels" in output_mgmt
239+
assert "emailConfiguration" in output_mgmt
240+
241+
# Email configuration
242+
email_config = output_mgmt["emailConfiguration"]
243+
assert email_config["to"] == ["finance@company.com", "accounting@company.com"]
244+
assert email_config["cc"] == ["manager@company.com"]
245+
assert email_config["emailNotificationTemplateKey"] == "PO_APPROVAL_NOTIFICATION"
246+
assert email_config["emailTemplateLanguage"] == "en"
247+
248+
@pytest.mark.asyncio
249+
async def test_send_email_with_mcp_multiple_recipients(self, email_client, sample_business_document, mock_mcp_tool):
250+
"""Test MCP email sending with multiple recipients."""
251+
recipients = [
252+
"finance@company.com",
253+
"accounting@company.com",
254+
"manager@company.com"
255+
]
256+
257+
result = await email_client.send_email_with_mcp(
258+
tool_name="send_output_request",
259+
notification_template_key="PO_APPROVAL_NOTIFICATION",
260+
to_emails=recipients,
261+
business_document=sample_business_document,
262+
mcp_tool=mock_mcp_tool
263+
)
264+
265+
assert result is not None
266+
mock_mcp_tool.ainvoke.assert_called_once()
267+
268+
# Verify all recipients were included
269+
call_args = mock_mcp_tool.ainvoke.call_args[0][0]
270+
email_config = call_args["body"]["data"]["outputManagement"]["emailConfiguration"]
271+
assert len(email_config["to"]) == 3
272+
assert set(email_config["to"]) == set(recipients)
273+
274+
@pytest.mark.asyncio
275+
async def test_send_email_with_mcp_complex_business_document(self, email_client, mock_mcp_tool):
276+
"""Test MCP email with complex nested business document."""
277+
complex_doc = {
278+
"Invoice": {
279+
"invoiceNumber": "INV-2024-001",
280+
"customer": {
281+
"id": "CUST-123",
282+
"name": "ACME Corporation",
283+
"address": {
284+
"street": "123 Main St",
285+
"city": "New York",
286+
"country": "USA"
287+
}
288+
},
289+
"lineItems": [
290+
{
291+
"itemId": "ITEM-001",
292+
"description": "Product A",
293+
"quantity": 10,
294+
"unitPrice": 100.00,
295+
"total": 1000.00
296+
},
297+
{
298+
"itemId": "ITEM-002",
299+
"description": "Product B",
300+
"quantity": 5,
301+
"unitPrice": 200.00,
302+
"total": 1000.00
303+
}
304+
],
305+
"subtotal": 2000.00,
306+
"tax": 200.00,
307+
"total": 2200.00
308+
}
309+
}
310+
311+
result = await email_client.send_email_with_mcp(
312+
tool_name="send_output_request",
313+
notification_template_key="INVOICE_NOTIFICATION",
314+
to_emails=["billing@customer.com"],
315+
business_document=complex_doc,
316+
mcp_tool=mock_mcp_tool
317+
)
318+
319+
assert result is not None
320+
mock_mcp_tool.ainvoke.assert_called_once()
321+
322+
# Verify business document was preserved
323+
call_args = mock_mcp_tool.ainvoke.call_args[0][0]
324+
business_doc = call_args["body"]["data"]["businessDocument"]
325+
assert "Invoice" in business_doc
326+
assert business_doc["Invoice"]["invoiceNumber"] == "INV-2024-001"
327+
assert len(business_doc["Invoice"]["lineItems"]) == 2
328+
329+
@pytest.mark.asyncio
330+
async def test_send_email_with_mcp_unique_trace_ids(self, email_client, sample_business_document, mock_mcp_tool):
331+
"""Test that each invocation generates unique trace IDs."""
332+
# Call the method twice
333+
await email_client.send_email_with_mcp(
334+
tool_name="send_output_request",
335+
notification_template_key="PO_APPROVAL_NOTIFICATION",
336+
to_emails=["finance@company.com"],
337+
business_document=sample_business_document,
338+
mcp_tool=mock_mcp_tool
339+
)
340+
341+
await email_client.send_email_with_mcp(
342+
tool_name="send_output_request",
343+
notification_template_key="PO_APPROVAL_NOTIFICATION",
344+
to_emails=["finance@company.com"],
345+
business_document=sample_business_document,
346+
mcp_tool=mock_mcp_tool
347+
)
348+
349+
# Verify two calls were made
350+
assert mock_mcp_tool.ainvoke.call_count == 2
351+
352+
# Get trace IDs from both calls
353+
call1_args = mock_mcp_tool.ainvoke.call_args_list[0][0][0]
354+
call2_args = mock_mcp_tool.ainvoke.call_args_list[1][0][0]
355+
356+
traceparent1 = call1_args["traceparent"]
357+
traceparent2 = call2_args["traceparent"]
358+
359+
# Verify they are different
360+
assert traceparent1 != traceparent2
361+
362+
@pytest.mark.asyncio
363+
async def test_send_email_with_mcp_no_optional_params(self, email_client, sample_business_document, mock_mcp_tool):
364+
"""Test MCP email with only required parameters."""
365+
result = await email_client.send_email_with_mcp(
366+
tool_name="send_output_request",
367+
notification_template_key="PO_APPROVAL_NOTIFICATION",
368+
to_emails=["finance@company.com"],
369+
business_document=sample_business_document,
370+
mcp_tool=mock_mcp_tool
371+
)
372+
373+
assert result is not None
374+
mock_mcp_tool.ainvoke.assert_called_once()
375+
376+
# Verify optional fields are not present or None
377+
call_args = mock_mcp_tool.ainvoke.call_args[0][0]
378+
email_config = call_args["body"]["data"]["outputManagement"]["emailConfiguration"]
379+
380+
# CC should not be present when not provided
381+
assert "cc" not in email_config or email_config.get("cc") is None
382+
383+
# Attachment should not be present when not provided
384+
assert "attachment" not in email_config or email_config.get("attachment") is None

0 commit comments

Comments
 (0)