import pytest
from CommonServerPython import *
from freezegun import freeze_time
from SendEmailReply import (
get_unique_code,
apply_direction,
process_directions,
ensure_markdown_tables_have_spacing,
replace_atlassian_tags,
)
def util_open_file(path):
with open(path) as f:
return f.read()
def util_load_json(path):
with open(path) as f:
return json.loads(f.read())
@pytest.mark.parametrize(
"inputted_body, expected_body",
[
pytest.param(
'Formatted HTML message.\r\n',
'Formatted HTML message.\r\n\r\nTest email signature.\r\n',
id="HTML Body",
),
pytest.param(
"Plain text message.
",
"Plain text message.
\r\nTest email signature.\r\n",
id="Plain Text Body",
),
],
)
def test_append_email_signature(mocker, inputted_body: str, expected_body: str):
"""
Given
- Email signature stored in XSOAR List
When
- List content is returned correctly
Then
- Validate that the returned message includes the appended signature content
"""
from SendEmailReply import append_email_signature
signature_list = util_load_json("test_data/getList_signature_success.json")
mocker.patch.object(demisto, "executeCommand", return_value=signature_list)
email_body = append_email_signature(inputted_body)
assert email_body == expected_body
def test_append_email_signature_fails(mocker):
"""
Given
- Email signature stored in XSOAR List
When
- List content results in error when fetched
Then
- Validate that the returned message remains unchanged
- Validate that a debug message is saved indicating the list couldn't be fetched
"""
from SendEmailReply import append_email_signature
get_list_error_response = util_load_json("test_data/getList_signature_error.json")
mocker.patch.object(demisto, "executeCommand", return_value=get_list_error_response)
debug_mocker = mocker.patch.object(demisto, "debug")
html_body = append_email_signature("Simple HTML message.\r\n")
debug_mocker_call_args = debug_mocker.call_args
assert (
debug_mocker_call_args.args[0] == "Error occurred while trying to load the `XSOAR - Email Communication "
"Signature` list. No signature added to email. Error: Item not found (8)."
)
assert html_body == "Simple HTML message.\r\n"
@pytest.mark.parametrize(
"email_cc, email_bcc, expected_result",
[
("", "", "Mail sent successfully. To: test1@gmail.com,test2@gmail.com"),
("cc_user@company.com", "", "Mail sent successfully. To: test1@gmail.com,test2@gmail.com Cc: cc_user@company.com"),
(
"cc_user@company.com",
"bcc_user@company.com",
"Mail sent successfully. To: test1@gmail.com,test2@gmail.com Cc: cc_user@company.com Bcc: bcc_user@company.com",
),
],
)
def test_validate_email_sent(email_cc, email_bcc, expected_result, mocker):
"""
Given
- Raw response of an email reply.
When
- The result is a successful reply with email recipients, cc, and bcc.
Then
- Validate that the successful message is returned.
"""
from SendEmailReply import validate_email_sent
email_reply_response = util_load_json("test_data/email_reply.json")
mocker.patch("SendEmailReply.execute_reply_mail", return_value=email_reply_response)
result = validate_email_sent(
"123",
"email_subject",
False,
"test1@gmail.com,test2@gmail.com",
"reply_body",
"html",
"test.onmicrosoft.com",
email_cc,
email_bcc,
"reply_html_body",
{},
"item_id",
"12345678",
"test.onmicrosoft.com",
"test.from.mail",
)
assert result == expected_result
def test_validate_email_sent_fails(mocker):
"""
Given -
a random error message which is returned from reply-mail executed command.
When -
executing the 'send_reply' function
Then -
an error message would be returned.
"""
from SendEmailReply import validate_email_sent
False, get_error(util_load_json("test_data/reply_mail_error.json"))
mocker.patch("SendEmailReply.execute_command", side_effect=Exception())
mocker.patch.object(demisto, "debug")
return_error_mock = mocker.patch("SendEmailReply.return_error")
with pytest.raises(Exception):
validate_email_sent("", "", False, "", "", "html", "", "", "", "", {}, "", "", "", "")
assert return_error_mock.call_count == 1
assert (
return_error_mock.call_args[0][0]
== 'SetIncident Failed."emailsubject" field was not updated with <> value for incident: '
)
@pytest.mark.parametrize(
"test_args, expected_response",
[
(
(
1,
"Email Subject",
False,
"end_user@company.com",
"Reply body.",
"html",
"soc_sender@company.com",
"cc_user@company.com",
"bcc_user@company.com",
"Reply body. Email Subject",
"cc": "cc_user@company.com",
"from": "from_mail",
"bcc": "bcc_user@company.com",
"htmlBody": "Reply body.Reply body. Email Subject",
"cc": "cc_user@company.com",
"bcc": "bcc_user@company.com",
"htmlBody": "Reply body.Reply body. [2] Email Subject",
"cc": "cc_user@company.com",
"from": "from_mail",
"bcc": "bcc_user@company.com",
"htmlBody": "Reply body.Reply body. [2] Email Subject",
"cc": "cc_user@company.com",
"bcc": "bcc_user@company.com",
"htmlBody": "Reply body."]',
"test123@gmail.com",
"avishai@demistodev.onmicrosoft.com",
"test@test.com",
{"test123@gmail.com", "avishai@demistodev.onmicrosoft.com"},
),
# Only mailbox is configured, should be removed.
(
'["avishai@demistodev.onmicrosoft.com", "test test <\'test@test.com\'>"]',
"test123@gmail.com",
"",
"test@test.com",
{"test123@gmail.com", "avishai@demistodev.onmicrosoft.com"},
),
# Only service mail is configured, should be removed.
(
'["avishai@demistodev.onmicrosoft.com", "test1@gmail.com"]',
"test123@gmail.com",
"avishai@demistodev.onmicrosoft.com",
"",
{"test123@gmail.com", "test1@gmail.com"},
),
# Neither service mail nor mailbox is configured, make sure nothing is removed.
(
'["avishai@demistodev.onmicrosoft.com", "test1@gmail.com"]',
"test123@gmail.com",
"",
"",
{"test123@gmail.com", "test1@gmail.com", "avishai@demistodev.onmicrosoft.com"},
),
# Mailbox is configured with different capitalization, but should still be removed.
(
'["avishai@demistodev.onmicrosoft.com", "test1@gmail.com"]',
"test123@gmail.com",
"",
"AVISHAI@DEMISTODEV.ONMICROSOFT.COM",
{"test123@gmail.com", "test1@gmail.com"},
),
# Service mail is configured with different capitalization, but should still be removed.
(
'["avishai@demistodev.onmicrosoft.com", "test1@gmail.com"]',
"test123@gmail.com",
"AVISHAI@DEMISTODEV.ONMICROSOFT.COM",
"",
{"test123@gmail.com", "test1@gmail.com"},
),
]
@pytest.mark.parametrize("email_to, email_from, service_mail, mailbox, excepted", GET_EMAIL_RECIPIENTS)
def test_get_email_recipients(email_to, email_from, service_mail, mailbox, excepted):
"""Unit test
Given
- Single email recipient, single email author, service mail and mailbox.
- Multiple email recipients, single email author, service mail and mailbox.
When
- Getting the email recipients.
Then
- validate that the correct email recipients are returned.
"""
from SendEmailReply import get_email_recipients
result = set(get_email_recipients(email_to, email_from, service_mail, mailbox).split(","))
assert result == excepted
@pytest.mark.parametrize(
"list_response, expected_result",
[
(util_load_json("test_data/getList_querywindow_success.json"), "success"),
(util_load_json("test_data/getList_querywindow_error.json"), "fail"),
],
)
def test_get_query_window(list_response, expected_result, mocker):
"""
Unit Test Scenario 1 - List exists
Given
- Query window value stored in XSOAR List
When
- List content is returned successfully
Then
- Validate that the function returns the correct window based on the list response
Unit Test Scenario 2 - List retrieval fails
Given
- Query window value stored in XSOAR List
When
- List retrieval results in an error
Then
- Validate that the function returns the default '60 days' window
- Validate that a debug message is saved indicating the list couldn't be fetched
"""
from SendEmailReply import get_query_window
mocker.patch.object(demisto, "executeCommand", return_value=list_response)
debug_mocker = mocker.patch.object(demisto, "debug")
result = get_query_window()
debug_mocker_call_args = debug_mocker.call_args
if expected_result == "success":
assert result == "90 days"
elif expected_result == "fail":
assert result == "60 days"
assert (
debug_mocker_call_args.args[0] == "Error occurred while trying to load the `XSOAR - Email "
"Communication Days To Query` list. Using the default query time - "
"60 days"
)
@pytest.mark.parametrize(
"notes, attachments, expected_results",
[
(
[{"Metadata": {"user": "DBot"}, "Contents": "note1"}, {"Metadata": {"user": "DBot"}, "Contents": "note2"}],
[{"name": "attachment1.png"}, {"name": "attachment2.png"}],
"DBot: \n\nnote1\n\nDBot: \n\nnote2\n\nAttachments: ['attachment1.png', 'attachment2.png']\n\n",
),
(
[{"Metadata": {"user": "DBot"}, "Contents": "note1"}, {"Metadata": {"user": "DBot"}, "Contents": "note2"}],
[],
"DBot: \n\nnote1\n\nDBot: \n\nnote2\n\n",
),
(
[{"Metadata": {"user": "DBot"}, "Contents": "note1"}, {"Metadata": {"user": "DBot"}, "Contents": "note2"}],
"[]",
"DBot: \n\nnote1\n\nDBot: \n\nnote2\n\n",
),
],
)
def test_get_reply_body(mocker, notes, attachments, expected_results):
"""Unit test
Given
- List of notes and list of attachments.
When
- Getting the email reply body.
Then
- validate that the correct reply is returned.
"""
import CommonServerPython
from SendEmailReply import get_reply_body
mocker.patch.object(demisto, "executeCommand", return_value=[{"EntryContext": {"replyhtmlbody": ""}, "Type": ""}])
mocker.patch.object(CommonServerPython, "dict_safe_get", return_value=None)
mocker.patch.object(CommonServerPython, "is_error", return_value=False)
reply_body = get_reply_body(notes=notes, incident_id="1", attachments=attachments)[0]
assert reply_body == expected_results
def test_create_file_data_json():
"""Unit test
Given
- Raw response of an attachment in email reply.
When
- There is an attachment in the email reply.
Then
- Validate that the file data is in the right json format.
"""
from SendEmailReply import create_file_data_json
attachment_response = util_load_json("test_data/attachment_example.json")
expected_result = util_open_file("test_data/file_data.txt")
result = create_file_data_json(attachment_response, "attachment")
assert result == expected_result
@pytest.mark.parametrize(
"current_cc, additional_cc, excepted",
[
(
"test1@gmail.com, test2@gmail.com",
"test3@gmail.com,test4@gmail.com",
"test1@gmail.com,test2@gmail.com,test3@gmail.com,test4@gmail.com",
),
("test1@gmail.com", "", "test1@gmail.com"),
("", "", ""),
],
)
def test_get_email_cc(current_cc, additional_cc, excepted):
"""Unit test
Given
- multiple current email cc and multiple additional email cc.
- single current email cc and empty additional email cc.
- empty current email cc and empty additional email cc.
When
- Getting email cc.
Then
- validate that the correct email cc are being returned.
"""
from SendEmailReply import get_email_cc
result = get_email_cc(current_cc, additional_cc)
assert result == excepted
def test_get_email_threads(mocker):
"""Unit Test
Given
- Function is called to fetch email threads from current incident
When
- Context contains email threads
Then
- Validate that function returns email thread data
"""
import SendEmailReply
from SendEmailReply import get_email_threads
email_threads = util_load_json("test_data/email_threads.json")
mocker.patch.object(demisto, "executeCommand")
mocker.patch.object(SendEmailReply, "dict_safe_get", return_value=email_threads)
result = get_email_threads("1")
assert result == email_threads
@pytest.mark.parametrize(
"email_code, email_threads, scenario",
[
("69433507", util_load_json("test_data/email_threads.json"), "thread_found"),
("123", util_load_json("test_data/email_threads.json"), "thread_notfound"),
(
"69433507",
[
{
"EmailCommsThreadId": "69433507",
"EmailCommsThreadNumber": "0",
"EmailCC": "cc_user@company.com",
"EmailBCC": "bcc_user@company.com",
"EmailBody": "Email body.",
"EmailFrom": "soc_sender@company.com",
"EmailHTML": "Email body.",
"MessageID": "5",
"EmailReceived": "soc_sender@company.com",
"EmailReplyTo": "soc_sender@company.com",
"EmailSubject": "Email Subject",
"EmailTo": "end_user@company.com",
"EmailAttachments": "None",
"MessageDirection": "outbound",
"MessageTime": "2022-02-04T20:56:53UTC",
}
],
"thread_found",
),
],
)
def test_create_thread_context(email_code, email_threads, scenario, mocker):
"""Unit test
Given:
- all required function arguments are provided
When:
- creating new context entry to store email thread data
Then
- validate that function calls appendContext() with all arguments and data needed to properly create
the required context entry
"""
import SendEmailReply
from SendEmailReply import create_thread_context
# Mock function to get current time string to match the expected result
mocker.patch(
"SendEmailReply.get_current_time_in_timezone",
return_value="2022-02-04T20:58:20UTC",
)
mocker.patch.object(SendEmailReply, "get_email_threads", return_value=email_threads)
append_context_mocker = mocker.patch.object(SendEmailReply, "appendContext", return_value=True)
create_thread_context(
email_code,
"cc_user@company.com",
"bcc_user@company.com",
"Email body.",
"soc_sender@company.com",
"body>",
"10",
"soc_sender@company.com",
"soc_sender@company.com",
"Email Subject",
"end_user@company.com",
"123",
"",
time_zone="UTC",
)
call_args = append_context_mocker.call_args
if scenario == "thread_found":
expected = {
"EmailCommsThreadId": "69433507",
"EmailCommsThreadNumber": "0",
"EmailCC": "cc_user@company.com",
"EmailBCC": "bcc_user@company.com",
"EmailBody": "Email body.",
"EmailFrom": "soc_sender@company.com",
"EmailHTML": "body>",
"MessageID": "10",
"EmailReceived": "soc_sender@company.com",
"EmailReplyTo": "soc_sender@company.com",
"EmailSubject": "Email Subject",
"EmailTo": "end_user@company.com",
"EmailAttachments": "",
"MessageDirection": "outbound",
"MessageTime": "2022-02-04T20:58:20UTC",
}
assert call_args.args[1] == expected
elif scenario == "thread_notfound":
expected = {
"EmailCommsThreadId": "123",
"EmailCommsThreadNumber": "2",
"EmailCC": "cc_user@company.com",
"EmailBCC": "bcc_user@company.com",
"EmailBody": "Email body.",
"EmailFrom": "soc_sender@company.com",
"EmailHTML": "body>",
"MessageID": "10",
"EmailReceived": "soc_sender@company.com",
"EmailReplyTo": "soc_sender@company.com",
"EmailSubject": "Email Subject",
"EmailTo": "end_user@company.com",
"EmailAttachments": "",
"MessageDirection": "outbound",
"MessageTime": "2022-02-04T20:58:20UTC",
}
assert call_args.args[1] == expected
@pytest.mark.parametrize(
"test_args, expected_result, expected_message",
[
(
( # Test arguments for send_new_email
1,
"Email Subject",
False,
"end_user@company.com",
"Email Body",
"soc_sender@company.com",
"",
"",
"Email Body", # email_html_body
"html",
[],
"12345678",
"soc_sender@company.com",
"attachment.txt",
"Email Body", # context_html_body
"from.mail",
"UTC",
),
( # Expected result for send_new_mail_request
1,
"Email Subject",
False,
"end_user@company.com",
"Email Body",
"soc_sender@company.com",
"",
"",
"Email Body + Signature", # Updated email_html_body
"html",
[],
"attachment.txt",
"12345678",
"soc_sender@company.com",
"Email Body + Signature", # Updated context_html_body
"from.mail",
"UTC",
),
"Mail sent successfully. To: end_user@company.com",
),
( # Test arguments for send_new_email
(
1,
"Email Subject",
False,
"end_user@company.com",
"Email Body",
"soc_sender@company.com",
"cc_user@company.com",
"",
"Email Body",
"html",
[],
"12345678",
"soc_sender@company.com",
"attachment.txt",
"Email Body",
"from.mail",
"UTC",
),
( # Expected result for send_new_mail_request
1,
"Email Subject",
False,
"end_user@company.com",
"Email Body",
"soc_sender@company.com",
"cc_user@company.com",
"",
"Email Body + Signature",
"html",
[],
"attachment.txt",
"12345678",
"soc_sender@company.com",
"Email Body + Signature",
"from.mail",
"UTC",
),
"Mail sent successfully. To: end_user@company.com Cc: cc_user@company.com",
),
( # Test arguments for send_new_email
(
1,
"Email Subject",
False,
"end_user@company.com",
"Email Body",
"soc_sender@company.com",
"",
"bcc_user@company.com",
"Email Body",
"html",
[],
"12345678",
"soc_sender@company.com",
"attachment.txt",
"Email Body",
"from_mail",
"UTC",
),
( # Expected result for send_new_mail_request
1,
"Email Subject",
False,
"end_user@company.com",
"Email Body",
"soc_sender@company.com",
"",
"bcc_user@company.com",
"Email Body + Signature",
"html",
[],
"attachment.txt",
"12345678",
"soc_sender@company.com",
"Email Body + Signature",
"from_mail",
"UTC",
),
"Mail sent successfully. To: end_user@company.com Bcc: bcc_user@company.com",
),
],
)
def test_send_new_email(test_args, expected_result, expected_message, mocker):
"""Unit test
Given:
- All required function arguments are provided
When:
- Appending email signature and calling send_new_mail_request function
Then
- Validate that the email signature is appended to 'email_html_body' and arguments are correctly
passed to 'send_new_mail_request'
"""
import SendEmailReply
from SendEmailReply import send_new_email
mocker.patch.object(SendEmailReply, "append_email_signature", return_value="Email Body + Signature")
send_new_mail_request_mocker = mocker.patch.object(
SendEmailReply, "send_new_mail_request", return_value=[{"Contents": "Success", "Type": 1}]
)
message = send_new_email(*test_args)
send_new_mail_request_args = send_new_mail_request_mocker.call_args
assert message == expected_message
assert send_new_mail_request_args.args == expected_result
@pytest.mark.parametrize(
"email_selected_thread, email_thread, expected_result",
[
("0", util_load_json("test_data/email_threads.json")[0], "success"),
("42", util_load_json("test_data/email_threads.json")[0], "fail"),
],
)
def test_resend_first_contact(email_selected_thread, email_thread, expected_result, mocker):
"""Unit Test
Given
- function is called to re-send a first-contact email message
When
- All input arguments are set
Then
- Validate that 'send_new_email' is called if selected thread is found
- Validate that an error is returned if selected thread is not found
"""
import SendEmailReply
from SendEmailReply import resend_first_contact
mocker.patch.object(SendEmailReply, "get_entry_id_list", return_value=["5", "10"])
mocker.patch.object(
SendEmailReply,
"format_body",
return_value=("Resending email.", "Resending email."),
)
mocker.patch.object(SendEmailReply, "get_email_cc", return_value="")
send_new_email_mocker = mocker.patch.object(SendEmailReply, "send_new_email", return_value=True)
return_error_mocker = mocker.patch.object(SendEmailReply, "return_error", return_value=True)
result = resend_first_contact(
email_selected_thread,
email_thread,
1,
"",
"",
"Resending email.",
"html",
"",
"",
"soc_sender@company.com",
"soc_sender@company.com",
"",
False,
"",
time_zone="UTC",
)
send_new_email_args = send_new_email_mocker.call_args
return_error_args = return_error_mocker.call_args
if expected_result == "success":
assert result is True
assert send_new_email_args.args == (
1,
"<69433507> Test Email 2",
False,
"end_user@company.com",
"Resending email.",
"soc_sender@company.com",
"",
"",
"Resending email.",
"html",
["5", "10"],
"69433507",
"soc_sender@company.com",
"",
"Resending email.",
"",
"UTC",
)
if expected_result == "fail":
assert return_error_args.args[0] == (
"The selected Thread Number to respond to (42) does not exist. Please choose a valid Thread Number and re-try."
)
@pytest.mark.parametrize("email_code", ["12345678", ""])
def test_single_thread_reply(email_code, mocker):
"""Unit Test
Given
- function is called to send a new email message
When
- All input arguments are correctly set
Then
- Validate that 'validate_email_sent' function is called and supplied with a correctly formatted set of mail content
- Validate that if no email_code is provided, 'get_unique_code' is called to generate one
"""
def get_reply_body_side_effect(notes, incident_id, attachments, reputation_calc_async): # noqa
return "Email body.", "Email body.", "Email body."
import SendEmailReply
from SendEmailReply import single_thread_reply
mocker.patch.object(SendEmailReply, "append_email_signature", side_effect=lambda body: body + " - signature")
mocker.patch.object(SendEmailReply, "get_unique_code", return_value="12345678")
execute_command_mocker = mocker.patch.object(demisto, "executeCommand", return_value=True)
mocker.patch.object(SendEmailReply, "get_entry_id_list", return_value=["5", "10"])
mocker.patch.object(SendEmailReply, "get_email_cc", return_value="")
mocker.patch.object(SendEmailReply, "get_reply_body", side_effect=get_reply_body_side_effect)
validate_email_sent_mocker = mocker.patch.object(SendEmailReply, "validate_email_sent", return_value=True)
single_thread_reply(
email_code,
1,
"cc_user@company.com",
"",
["5"],
"html",
"",
"",
"Email Subject",
False,
"end_user@company.com",
"soc_sender@company.com",
10,
"soc_sender@company.com",
"",
)
validate_mail_sent_call_args = validate_email_sent_mocker.call_args
execute_command_call_args = execute_command_mocker.call_args
assert validate_mail_sent_call_args.args == (
1,
"Email Subject",
False,
"end_user@company.com",
"Email body.",
"html",
"soc_sender@company.com",
"",
"",
"Email body. - signature", # The expected value is now a signed body
["5", "10"],
10,
"12345678",
"soc_sender@company.com",
"",
)
if not email_code:
assert execute_command_call_args.args == ("setIncident", {"id": 1, "customFields": {"emailgeneratedcode": "12345678"}})
@pytest.mark.parametrize(
"test_args, expected_result",
[
(
(
1,
"Email Subject",
False,
"end_user@company.com",
"Message body.",
"soc_sender@company.com",
"cc_user@company.com",
"bcc_user@company.com",
"Reply body. Email Subject",
"cc": "cc_user@company.com",
"bcc": "bcc_user@company.com",
"htmlBody": "Reply body.Reply body. Email Subject",
"cc": "cc_user@company.com",
"bcc": "bcc_user@company.com",
"htmlBody": "Reply body.Some HTML", "Some HTML"))
send_new_email_mocker = mocker.patch.object(SendEmailReply, "send_new_email", return_value=True)
reset_fields_mocker = mocker.patch.object(SendEmailReply, "reset_fields", return_value=True)
if scenario == "required_fields_missing":
# Test Scenario 1
expected = (
"The following required fields have not been set. Please set them and try again. "
"['New Email Subject', 'New Email Recipients', 'New Email Body']"
)
multi_thread_new(
"",
False,
"",
"",
"html",
1,
"12345678",
"",
"",
"soc_sender@company.com",
"cc_user@company.com",
"bcc_user@company.com",
"soc_sender@company.com",
"",
"",
time_zone="UTC",
)
call_args = return_error_mocker.call_args
assert call_args.args[0] == expected
if scenario == "no_codes_present":
# Test Scenario 2
multi_thread_new(
"New Subject",
False,
"end_user@company.com",
"Email Body",
"html",
1,
"",
"",
"",
"soc_sender@company.com",
"cc_user@company.com",
"bcc_user@company.com",
"soc_sender@company.com",
"",
"",
time_zone="UTC",
)
set_incident_call_args = set_incident_mocker.call_args
send_new_email_mocker_args = send_new_email_mocker.call_args
assert set_incident_call_args.args[1] == {"id": 1, "customFields": {"emailgeneratedcodes": "87654321"}}
valid_args = (
1,
"New Subject",
False,
"end_user@company.com",
"Email Body",
"soc_sender@company.com",
"cc_user@company.com",
"bcc_user@company.com",
"Some HTML",
"html",
[],
"87654321",
"soc_sender@company.com",
"",
"Some HTML",
"",
"UTC",
)
assert send_new_email_mocker_args.args == valid_args
assert reset_fields_mocker.called is True
if scenario == "codes_present":
# Test Scenario 3
multi_thread_new(
"New Subject",
False,
"end_user@company.com",
"Email Body",
"html",
1,
"12345678",
"",
"",
"soc_sender@company.com",
"cc_user@company.com",
"bcc_user@company.com",
"soc_sender@company.com",
"",
"",
time_zone="UTC",
)
set_incident_call_args = set_incident_mocker.call_args
send_new_email_mocker_args = send_new_email_mocker.call_args
assert set_incident_call_args.args[1] == {"id": 1, "customFields": {"emailgeneratedcodes": "12345678,87654321"}}
valid_args = (
1,
"New Subject",
False,
"end_user@company.com",
"Email Body",
"soc_sender@company.com",
"cc_user@company.com",
"bcc_user@company.com",
"Some HTML",
"html",
[],
"87654321",
"soc_sender@company.com",
"",
"Some HTML",
"",
"UTC",
)
assert send_new_email_mocker_args.args == valid_args
assert reset_fields_mocker.called is True
def test_collect_thread_details():
"""
Unit test scenario
Given
- Function is called to gather details from existing email thread
When
- Multiple emails are present on the thread
Then
- Validate that the function returns the correct email details for use in an email reply
"""
from SendEmailReply import collect_thread_details
expected = (
True,
"AAMkAGRcOGZlZTEzLTkyZGDtNGJkNy1iOWMxLYM0NTAwODZhZjlxNABGAAAAAAAP2ksrJ8icRL4Zhadm7iVXBwAkkBJXBb0"
"sRJWC0zdXEMqsAAAAAAEMAAAkkBJFBb0fRJWC0zdXEMqsABApcWVYAAA=",
False,
"87692312",
"Re: <87692312> Test Email 4",
"end_user@company.com, soc_sender@company.com",
"soc_sender@company.com",
"",
"",
3,
)
actual = collect_thread_details(util_load_json("test_data/email_threads.json"), "1")
assert actual == expected
@pytest.mark.parametrize("scenario", ["single_outbound", "multiple_outbound", "replies_present"])
def test_multi_thread_reply(scenario, mocker):
"""
Unit test scenario 1 - Resend First-Contact - Single Outbound
Given
- Function is called to send a reply to an existing email thread
When
- The selected thread contains only an outbound message (no replies from recipients)
Then
- Validate that the function calls 'resend_first_contact' to send the message as a new outbound email
- Validate that 'first_contact_resent' is set to True
Unit test scenario 2 - Resend First-Contact - Multiple Outbound
Given
- Function is called to send a reply to an existing email thread
When
- The selected thread contains multiple outbound messages (no replies from recipients)
Then
- Validate that the function calls 'resend_first_contact' to send the message as a new outbound email
- Validate that 'first_contact_resent' is set to True
Unit test scenario 3 - Replies Present
Given
- Function is called to send a reply to an existing email thread
When
- The selected thread contains both outbound and inbound messages (there are replies from recipients)
Then
- Validate that the function calls 'send_reply' to reply to existing email thread
- Validate that the function calls 'create_thread_context' with correct arguments
- Validate that 'first_contact_resent' is set to False
- Validate that 'reset_fields' is called
"""
import SendEmailReply
from SendEmailReply import multi_thread_reply
email_threads = util_load_json("test_data/email_threads.json")
resend_first_contact_mocker = mocker.patch.object(SendEmailReply, "resend_first_contact", return_value=True)
mocker.patch.object(SendEmailReply, "reset_fields", return_value=True)
mocker.patch.object(SendEmailReply, "return_results", return_value=True)
mocker.patch.object(
SendEmailReply,
"format_body",
return_value=("Email body", "Email body"),
)
mocker.patch.object(SendEmailReply, "append_email_signature", return_value="Email body+signature")
reset_fields_mocker = mocker.patch.object(SendEmailReply, "reset_fields", return_value=True)
validate_email_sent_mocker = mocker.patch.object(SendEmailReply, "validate_email_sent", return_value=True)
create_context_mocker = mocker.patch.object(SendEmailReply, "create_thread_context", return_value=True)
if scenario == "single_outbound":
# Return only a single email thread entry
mocker.patch.object(SendEmailReply, "get_email_threads", return_value=email_threads[0])
multi_thread_reply(
"Email body",
"html",
1,
0,
"",
"",
"cc_user@company.com",
"bcc_user@company.com",
"soc_sender@company.com",
"soc_sender@company.com",
"",
False,
"",
time_zone="UTC",
)
expected = (
0,
{
"EmailBCC": "",
"EmailBody": "Outbound test message from XSOAR to User.",
"EmailCC": "",
"EmailCommsThreadId": "69433507",
"EmailCommsThreadNumber": "0",
"EmailFrom": "soc_sender@company.com",
"EmailHTML": "Outbound test message from XSOAR to User.",
"EmailReceived": "",
"EmailReplyTo": "soc_sender@company.com",
"EmailSubject": "<69433507> Test Email 2",
"EmailTo": "end_user@company.com",
"EmailAttachments": "None",
"MessageDirection": "outbound",
"MessageID": "",
"MessageTime": "2022-02-04T20:56:53UTC",
},
1,
"",
"",
"Email body",
"html",
"cc_user@company.com",
"bcc_user@company.com",
"soc_sender@company.com",
"soc_sender@company.com",
"",
False,
"",
"UTC",
)
resend_first_contact_call_args = resend_first_contact_mocker.call_args
assert resend_first_contact_call_args.args == expected
assert reset_fields_mocker.called is True
elif scenario == "multiple_outbound":
# Return all email thread entries
mocker.patch.object(SendEmailReply, "get_email_threads", return_value=email_threads)
thread_details = (
True,
"AAMkAGRcOGZlZTEzLTkyZGDtNGJkNy1iOWMxLYM0NTAwODZhZjlxNABGAAAAAAAP2ksrJ8icRL4Zhadm7iVXBwAkkBJXBb0"
"sRJWC0zdXEMqsAAAAAAEMAAAkkBJFBb0fRJWC0zdXEMqsABApcWVYAAA=",
True,
"69433507",
"Re: <69433507> Test Email 2",
"end_user@company.com, soc_sender@company.com",
"soc_sender@company.com",
"",
"",
2,
)
mocker.patch.object(SendEmailReply, "collect_thread_details", return_value=thread_details)
# Execute the tested function
multi_thread_reply(
"Email body",
"html",
1,
0,
"",
"",
"cc_user@company.com",
"bcc_user@company.com",
"soc_sender@company.com",
"soc_sender@company.com",
"",
False,
"",
time_zone="UTC",
)
expected = (
0,
{
"EmailBCC": "",
"EmailBody": "Outbound test message from XSOAR to User.",
"EmailCC": "",
"EmailCommsThreadId": "87692312",
"EmailCommsThreadNumber": "1",
"EmailFrom": "soc_sender@company.com",
"EmailHTML": "Outbound test message from XSOAR to User.",
"EmailReceived": "",
"EmailReplyTo": "soc_sender@company.com",
"EmailSubject": "<87692312> Test Email 4",
"EmailTo": "end_user@company.com",
"EmailAttachments": "None",
"MessageDirection": "outbound",
"MessageID": "",
"MessageTime": "2022-02-04T20:56:53UTC",
},
1,
"",
"",
"Email body",
"html",
"cc_user@company.com",
"bcc_user@company.com",
"soc_sender@company.com",
"soc_sender@company.com",
"",
False,
"",
"UTC",
)
resend_first_contact_call_args = resend_first_contact_mocker.call_args
assert resend_first_contact_call_args.args == expected
assert reset_fields_mocker.called is True
elif scenario == "replies_present":
thread_details = (
True,
"AAMkAGRcOGZlZTEzLTkyZGDtNGJkNy1iOWMxLYM0NTAwODZhZjlxNABGAAAAAAAP2ksrJ8icRL4Zhadm7iVXBwAkkBJXBb0"
"sRJWC0zdXEMqsAAAAAAEMAAAkkBJFBb0fRJWC0zdXEMqsABApcWVYAAA=",
False,
"87692312",
"RE: <87692312> Test Email 4",
"end_user@company.com, soc_sender@company.com",
"soc_sender@company.com",
"",
"",
2,
)
mocker.patch.object(SendEmailReply, "collect_thread_details", return_value=thread_details)
# Return all email thread entries
mocker.patch.object(SendEmailReply, "get_email_threads", return_value=email_threads)
validate_email_sent_expected = (
1,
"<87692312> Test Email 4",
False,
"end_user@company.com",
"Email body",
"html",
"soc_sender@company.com",
"cc_user@company.com",
"bcc_user@company.com",
"Email body+signature",
[],
"AAMkAGRcOGZlZTEzLTkyZGDtNGJkNy1iOWMxLYM0NTAwODZhZjlxNABGAAAAAAAP2ksrJ8icRL4Zha"
"dm7iVXBwAkkBJXBb0sRJWC0zdXEMqsAAAAAAEMAAAkkBJFBb0fRJWC0zdXEMqsABApcWVYAAA=",
"87692312",
"soc_sender@company.com",
"",
)
create_context_expected = (
"87692312",
"cc_user@company.com",
"bcc_user@company.com",
"Email body",
"soc_sender@company.com",
"Email body+signature",
"",
"",
"soc_sender@company.com",
"<87692312> Test Email 4",
"end_user@company.com",
1,
"",
"UTC",
)
# Execute the tested function
multi_thread_reply(
"Email body",
"html",
1,
1,
"",
"",
"cc_user@company.com",
"bcc_user@company.com",
"soc_sender@company.com",
"soc_sender@company.com",
"",
False,
"",
time_zone="UTC",
)
validate_email_sent_call_args = validate_email_sent_mocker.call_args
create_context_call_args = create_context_mocker.call_args
assert validate_email_sent_call_args.args == validate_email_sent_expected
assert create_context_call_args.args == create_context_expected
assert reset_fields_mocker.called is True
@pytest.mark.parametrize("new_thread", ["n/a", "true", "false"])
def test_main(new_thread, mocker):
"""
Unit Test Scenario 1 new_thread = 'n/a'
Given
- Script is called to send an email message
When
- new_thread = 'n/a', indicating this script is being called from the Email Communication layout
Then
- Validate that function single_thread_reply is called with all correct arguments
Unit Test Scenario 1 new_thread = 'true'
Given
- Script is called to send an email message
When
- new_thread = 'true', indicating this script is being called from the Email Threads layout and the message
to be sent is a new first-contact email
Then
- Validate that function multi_thread_new is called with all correct arguments
Unit Test Scenario 1 new_thread = 'false'
Given
- Script is called to send an email message
When
- new_thread = 'false', indicating this script is being called from the Email Threads layout and the message
to be sent is a reply to an existing email chain
Then
- Validate that function multi_thread_reply is called with all correct arguments
"""
import SendEmailReply
from SendEmailReply import main
incident = util_load_json("test_data/incident_data.json")
input_args = {
"service_mail": "soc_sender@company.com",
"files": {},
"mail_sender_instance": "mail-sender-instance-1",
"new_thread": new_thread,
"bodyType": "html",
}
mocker.patch.object(demisto, "args", return_value=input_args)
mocker.patch.object(demisto, "incident", return_value=incident)
mocker.patch.object(SendEmailReply, "get_mailbox_from_incident_labels", return_value="")
mocker.patch.object(SendEmailReply, "get_email_recipients", return_value="end_user@company.com")
mocker.patch.object(demisto, "executeCommand", return_value="")
single_thread_reply_mocker = mocker.patch.object(SendEmailReply, "single_thread_reply", return_value=True)
multi_thread_new_mocker = mocker.patch.object(SendEmailReply, "multi_thread_new", return_value=True)
multi_thread_reply_mocker = mocker.patch.object(SendEmailReply, "multi_thread_reply", return_value=True)
main()
if new_thread == "n/a":
single_thread_reply_args = single_thread_reply_mocker.call_args
expected_args = (
"87654321",
"10",
"",
"test_cc@example.com",
"",
"html",
[],
{},
None,
False,
"end_user@company.com",
"soc_sender@company.com",
"123456",
"mail-sender-instance-1",
None,
False,
)
assert single_thread_reply_args.args == expected_args
elif new_thread == "true":
multi_thread_new_args = multi_thread_new_mocker.call_args
expected_args = (
"Test Email Subject.",
False,
"test_recipient@example.com",
"This is a test email.",
"html",
"10",
None,
{},
{},
"soc_sender@company.com",
"test_cc@example.com",
"test_bcc@example.com",
"mail-sender-instance-1",
"None",
None,
"UTC",
)
assert multi_thread_new_args.args == expected_args
elif new_thread == "false":
multi_thread_reply_args = multi_thread_reply_mocker.call_args
expected_args = (
"This is a test email.",
"html",
"10",
1,
{},
{},
"test_cc@example.com",
"test_bcc@example.com",
"soc_sender@company.com",
"mail-sender-instance-1",
"None",
False,
None,
"UTC",
)
assert multi_thread_reply_args.args == expected_args
# Parametrized test for happy path scenarios with various realistic markdown inputs
@pytest.mark.parametrize(
"input_md, expected_html, test_id",
[
# Test ID: #1 - Simple text conversion
("Hello, World!", "Hello, World!
", "simple_text"),
# Test ID: #2 - Header conversion
("# Header 1", "Header 1
", "header_conversion"),
# Test ID: #3 - Table conversion
(
"| Header1 | Header2 |\n| ------- | ------- |\n| Cell1 | Cell2 |",
'\n'
"\n\n| Header1 | \nHeader2 | \n
\n\n"
"\n\n| Cell1 | \nCell2 | \n
\n\n
",
"table_conversion",
),
# Test ID: #4 - Emphasis conversion using legacy syntax
("_italic_ **bold**", "italic bold
", "emphasis_conversion"),
# Test ID: #5 - List conversion
("- Item 1\n- Item 2", "", "list_conversion"),
# Test ID: #6 - New lines to
conversion
("Line 1\nLine 2", "Line 1
\nLine 2
", "newline_to_br_conversion"),
],
ids=lambda test_id: test_id,
)
def test_format_body_happy_path(input_md, expected_html, test_id):
# Act
from SendEmailReply import format_body
result = format_body(input_md)
# Assert
assert result[1] == expected_html, f"Test failed for {test_id}"
# Parametrized test for edge cases
@pytest.mark.parametrize(
"input_md, expected_html, test_id",
[
# Test ID: #1 - Empty string
("", ("", ""), "empty_string"),
# Test ID: #2 - Markdown with only special characters
("# $%^&*()", ("$%^&*()
", "$%^&*()
"), "special_characters_only"),
],
ids=lambda test_id: test_id,
)
def test_format_body_edge_cases(input_md, expected_html, test_id):
# Act
from SendEmailReply import format_body
result = format_body(input_md)
# Assert
assert result == expected_html, f"Test failed for {test_id}"
# Parametrized test for edge cases
@pytest.mark.parametrize(
"input_md, expected_html, test_id",
[
# Test ID: #1 - Demisto custom markdown underline syntax.
("+underline+", ("underline
", "underline
"), "underline"),
# Test ID: #2 - Demisto custom markdown strikethrough syntax.
("~~strikethrough~~", ("strikethrough
", "strikethrough
"), "strikethrough"),
],
ids=lambda test_id: test_id,
)
def test_demisto_custom_markdown_syntax(input_md, expected_html, test_id):
# Act
from SendEmailReply import format_body
result = format_body(input_md)
# Assert
assert result == expected_html, f"Test failed for {test_id}"
def test_format_body_raw_checkbox_true(mocker):
"""
Given
- An incident with the checkbox 'Send Body as Raw Text (No Markdown)' checked
When
- The format_body function is called with a markdown-formatted body
Then
- Validate that the body is returned as-is, without any markdown conversion
"""
from SendEmailReply import format_body
mock_incident = {"CustomFields": {"sendbodyasrawnomarkdown": True}}
mocker.patch("SendEmailReply.demisto.incident", return_value=mock_incident)
result = format_body("**bold**")
assert result == ("**bold**", "**bold**")
result = format_body("_using italic_")
assert result == ("_using italic_", "_using italic_")
@freeze_time("2024-02-22 10:00:00 UTC")
def test_get_unique_code_happy_path(mocker):
# Arrange
incident_id = "123"
max_tries = 1000
expected_code = "1231708596000000"
mocker.patch("SendEmailReply.demisto", return_value=None)
mocker.patch("SendEmailReply.get_incident_by_query", return_value=None)
# Act
code = get_unique_code(incident_id, max_tries)
# Assert
assert code == expected_code
@freeze_time("2024-02-22 10:00:00 UTC")
def test_get_unique_code_edge_cases(mocker):
# Arrange
incident_id = "1"
max_tries = 1000
expected_code = "0011708596000000"
mocker.patch("SendEmailReply.get_incident_by_query", return_value=None)
# Act
code = get_unique_code(incident_id, max_tries)
# Assert
assert code == expected_code
@freeze_time("2024-02-22 10:00:00 UTC")
def test_get_unique_code_error_case(mocker):
# Arrange
incident_id = "123"
max_tries = 0
mocker.patch("SendEmailReply.get_incident_by_query", return_value=[{"id": "existing_incident"}])
mocked_return_error = mocker.patch("SendEmailReply.return_error", side_effect=Exception("Error"))
# Act and Assert
with pytest.raises(Exception):
get_unique_code(incident_id, max_tries)
mocked_return_error.assert_called_once_with(
f"Failed to generate unique code for incident {incident_id} after {max_tries} tries"
)
def test_format_body(mocker):
from SendEmailReply import format_body
html_body = ""
mocker.patch("SendEmailReply.execute_command", return_value=(True, [{"FileID": "111"}]))
mocker.patch.object(demisto, "investigation", return_value={"id": "1234"})
open_mock = mocker.mock_open(read_data=b"some binary data")
mocker.patch("builtins.open", open_mock)
result = format_body(html_body)
expected_result = (
'
',
'
',
)
assert result == expected_result
def test_apply_direction():
"""
Test that apply_direction correctly transforms lines with direction tags
into the expected HTML div wrappers, and leaves other lines unchanged.
"""
# Center alignment
assert apply_direction("<-:->Centered text") == 'Centered text
'
# Leading spaces should be ignored for detecting tags
assert apply_direction(" <-:->Indented center") == 'Indented center
'
# Left-to-right direction
assert apply_direction("<--:>LTR text") == 'LTR text
'
# Right-to-left direction
assert apply_direction("<:-->RTL text") == 'RTL text
'
# No direction tag, returns original line
assert apply_direction("No special tag") == "No special tag"
# Tag-like but not at start after stripping
assert apply_direction(" some text <-:-> not a tag") == " some text <-:-> not a tag"
def test_process_directions():
"""
Test that process_directions applies apply_direction line-by-line on multiline text.
"""
md = """Line 1
<-:->Center
Some text
<--:>LTR line
<:-->RTL line"""
expected = (
"Line 1\n"
'Center
\n'
"Some text\n"
'LTR line
\n'
'RTL line
'
)
assert process_directions(md) == expected
def test_ensure_markdown_tables_have_spacing():
"""
Test that ensure_markdown_tables_have_spacing adds blank lines before and after markdown tables.
It should not add extra blank lines if already present.
"""
# Single table without blank lines
md = "Paragraph\n| Header |\n|--------|\n| Row |\nParagraph2"
expected = "Paragraph\n\n| Header |\n|--------|\n| Row |\n\nParagraph2"
assert ensure_markdown_tables_have_spacing(md) == expected
# Table already has blank lines
md2 = "Para\n\n| H |\n|---|\n| R |\n\nPara2"
expected2 = md2
assert ensure_markdown_tables_have_spacing(md2) == expected2
# Multiple tables with some blank lines missing
md3 = "| H1 |\n|----|\n| R1 |\nPara\n| H2 |\n|----|\n| R2 |"
expected3 = "| H1 |\n" "|----|\n" "| R1 |\n" "\n" "Para\n" "\n" "| H2 |\n" "|----|\n" "| R2 |"
assert ensure_markdown_tables_have_spacing(md3).rstrip() == expected3.rstrip()
def test_replace_atlassian_tags():
"""
Test that replace_atlassian_tags converts Atlassian color and background tags into HTML span tags.
"""
# Color tag
input_md = "{{color:#FF0000}}(red text)"
expected = 'red text'
assert replace_atlassian_tags(input_md) == expected
# Background tag
input_md2 = "{{background:#00FF00}}(green background)"
expected2 = 'green background'
assert replace_atlassian_tags(input_md2) == expected2
# Both tags in one string
input_md3 = "Text {{color:#123456}}(colored) and {{background:#654321}}(background)"
expected3 = 'Text colored and background'
assert replace_atlassian_tags(input_md3) == expected3
# No tags remain unchanged
input_md4 = "No tags here"
assert replace_atlassian_tags(input_md4) == input_md4
class TestGetReplyBodyXSSPrevention:
"""Tests for XSS prevention in attachment names within get_reply_body."""
def test_attachment_names_escaped_in_reply_body(self, mocker):
"""
Given
- A note with content and attachments containing XSS payloads in filenames
When
- get_reply_body is called
Then
- Validate that attachment names in the reply body are HTML-escaped
"""
from SendEmailReply import get_reply_body
notes = [{"Metadata": {"user": "admin"}, "Contents": "Test reply", "ID": "1"}]
malicious_attachments = [
{"name": ""},
{"name": "
"},
]
mocker.patch(
"SendEmailReply.execute_command",
side_effect=[
(True, [{"Contents": {"name": "Admin User"}}]), # getUserByUsername
(True, None), # core-api-post (entry/note)
(True, None), # addEntries
],
)
mocker.patch.object(demisto, "incident", return_value={"CustomFields": {"sendbodyasrawnomarkdown": False}})
reply_body, context_html, reply_html = get_reply_body(notes, "123", malicious_attachments)
# Attachment names must be escaped in the reply body
assert "<script>" in reply_body
assert "" not in reply_body
assert "<img src=x onerror=alert(1)>" in reply_body
assert "