import demistomock as demisto
import pytest
from CommonServerPython import * # noqa: F401
from requests import Response
from MicrosoftTeams import GraphPermissions as Perms
from freezegun import freeze_time
entryTypes["warning"] = 11
GRAPH_BASE_URL: str = "https://graph.microsoft.com"
BASE_URL = "https://graph.microsoft.com/v1.0/groups"
bot_id: str = "9bi5353b-md6a-4458-8321-e924af433amb"
tenant_id: str = "pbae9ao6-01ql-249o-5me3-4738p3e1m941"
team_id: str = "19:21f27jk08d1a487fa0f5467779619827@thread.skype"
channel_id: str = "19:4b6bed8d24574f6a9e436813cb2617d8@thread.tacv2"
team_aad_id: str = "7d8efdf8-0c5a-42e3-a489-5ef5c3fc7a2b"
team_name: str = "The-A-Team"
service_url: str = "https://smba.trafficmanager.net/emea"
mirrored_channels: list = [
{
"channel_id": "19:2cbad0d78c624400ef83a5750539998g@thread.skype",
"investigation_id": "1",
"mirror_type": "all",
"mirror_direction": "both",
"auto_close": "true",
"mirrored": True,
"channel_name": "incident-1",
},
{
"channel_id": "19:2cbad0d78c624400ef83a5750534448g@thread.skype",
"investigation_id": "10",
"mirror_type": "all",
"mirror_direction": "both",
"auto_close": "true",
"mirrored": True,
"channel_name": "incident-10",
},
]
team_members: list = [
{
"id": "29:1KZccCJRTxlPdHnwcKfxHAtYvPLIyHgkSLhFSnGXLGVFlnltovdZPmZAduPKQP6NrGqOcde7FXAF7uTZ_8FQOqg",
"objectId": "359d2c3c-162b-414c-b2eq-386461e5l050",
"name": "Bruce Willis",
"givenName": "Bruce",
"surname": "Willis",
"userPrincipalName": "bwillis@email.com",
"email": "bwillis@email.com",
"tenantId": tenant_id,
},
{
"id": "29:1pBMMC85IyjM3tr_MCZi7KW4pw4EULxLN4C7R_xoi3Wva_lOn3VTf7xJlCLK-r-pMumrmoz9agZxsSrCf7__u9R",
"objectId": "2826c1p7-bdb6-4529-b57d-2598me968631",
"name": "Denzel Washington",
"givenName": "Denzel",
"surname": "Washington",
"email": "DwashintoN@email.com",
"userPrincipalName": "DwashintoN@email.com",
"tenantId": tenant_id,
},
]
channel_members: dict = {
"@odata.context": "https://graph.microsoft.com/v1.0/$metadata#teams('2ab9c796-2902-45f8-b712-7c5a63cf41c4')/"
"channels('19%3A20bc1df46b1148e9b22539b83bc66809%40thread.skype')/members",
"@odata.count": 2,
"value": [
{
"@odata.type": "#microsoft.graph.aadUserConversationMember",
"id": "MmFiOWM3OTYtMjkwMi00NWY4LWI3MTItN2M1YTYzY2Y0MWM0IyNlZWY5Y2IzNi0wNmRlLTQ2OWItODdjZC03MGY0Y2JlMzJkMTQ=",
"roles": [],
"displayName": "Jane Doe",
"userId": "eef9cb36-06de-469b-87cd-70f4cbe32d14",
"email": "jdoe@teamsip.onmicrosoft.com",
"tenantId": tenant_id,
"visibleHistoryStartDateTime": "0001-01-01T00:00:00Z",
},
{
"@odata.type": "#microsoft.graph.aadUserConversationMember",
"id": "MmFiOWM3OTYtMjkwMi00NWY4LWI3MTItN2M1YTYzY2Y0MWM0IyNiMzI0NmY0NC1jMDkxLTQ2MjctOTZjNi0yNWIxOGZhMmM5MTA=",
"roles": ["owner"],
"displayName": "Ace John",
"userId": "b3246f44-c091-4627-96c6-25b18fa2c910",
"email": "ajohn@teamsip.onmicrosoft.com",
"tenantId": tenant_id,
"visibleHistoryStartDateTime": "0001-01-01T00:00:00Z",
},
],
}
integration_context: dict = {
"bot_name": "DemistoBot",
"service_url": service_url,
"tenant_id": tenant_id,
"teams": json.dumps(
[
{
"mirrored_channels": mirrored_channels,
"team_id": team_id,
"team_aad_id": team_aad_id,
"team_members": team_members,
"team_name": team_name,
}
]
),
}
CLIENT_CREDENTIALS_FLOW = "Client Credentials"
AUTHORIZATION_CODE_FLOW = "Authorization Code"
CREDENTIALS_TOKEN_PARAMS = "credentials_token_params"
AUTHCODE_TOKEN_PARAMS = "authcode_token_params"
ONEONONE_CHAT_ID = "19:09ddc990-3821-4ceb-8019-24d39998f93e_48d31887-5fad-4d73-a9f5-3c356e68a038@unq.gbl.spaces"
GROUP_CHAT_ID = "19:2da4c29f6d7041eca70b638b43d45437@thread.v2"
def util_load_json(path: str):
with open(path, encoding="utf-8") as f:
return json.loads(f.read())
@pytest.fixture(autouse=True)
def get_integration_context(mocker):
mocker.patch.object(demisto, "getIntegrationContext", return_value=integration_context)
@pytest.fixture(autouse=True)
def get_graph_access_token(requests_mock):
requests_mock.post(
f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token", json={"access_token": "token"}, status_code=200
)
@pytest.fixture(autouse=True)
def get_bot_access_token(requests_mock):
requests_mock.post("https://login.microsoftonline.com/botframework.com/oauth2/v2.0/token", json={"access_token": "token"})
def test_mentioned_users_to_entities():
from MicrosoftTeams import mentioned_users_to_entities
mentioned_users = ["Bruce Willis", "Denzel Washington"]
bruce_entity = {
"type": "mention",
"mentioned": {
"id": "29:1KZccCJRTxlPdHnwcKfxHAtYvPLIyHgkSLhFSnGXLGVFlnltovdZPmZAduPKQP6NrGqOcde7FXAF7uTZ_8FQOqg",
"name": "Bruce Willis",
},
"text": "@Bruce Willis",
}
denzel_entity = {
"type": "mention",
"mentioned": {
"id": "29:1pBMMC85IyjM3tr_MCZi7KW4pw4EULxLN4C7R_xoi3Wva_lOn3VTf7xJlCLK-r-pMumrmoz9agZxsSrCf7__u9R",
"name": "Denzel Washington",
},
"text": "@Denzel Washington",
}
assert mentioned_users_to_entities(mentioned_users, integration_context) == [bruce_entity, denzel_entity]
mentioned_users = ["Bruce Willis", "demisto"]
with pytest.raises(ValueError, match="Team member demisto was not found"):
mentioned_users_to_entities(mentioned_users, integration_context)
def test_process_mentioned_users_in_message():
from MicrosoftTeams import process_mentioned_users_in_message
raw_message = "@demisto dev; @demisto; a@demisto.com; a@demisto.com hi; @hi @wow;"
parsed_message = "@demisto dev @demisto a@demisto.com; a@demisto.com hi; @hi @wow"
users, message = process_mentioned_users_in_message(raw_message)
assert users == ["demisto dev", "demisto", "wow"]
assert message == parsed_message
def test_message_handler(mocker):
from MicrosoftTeams import message_handler
mocker.patch.object(demisto, "addEntry")
request_body: dict = {
"from": {
"id": "29:1KZccCJRTxlPdHnwcKfxHAtYvPLIyHgkSLhFSnGXLGVFlnltovdZPmZAduPKQP6NrGqOcde7FXAF7uTZ_8FQOqg",
"aadObjectId": "359d2c3c-162b-414c-b2eq-386461e5l050",
"name": "Bruce Willis",
}
}
channel_data: dict = {"channel": {"id": "19:2cbad0d78c624400ef83a5750539998g@thread.skype"}, "team": {"id": team_id}}
message_handler(integration_context, request_body, channel_data, "waz up")
assert demisto.addEntry.call_count == 1
add_entry_args = demisto.addEntry.call_args[1]
assert add_entry_args == {
"id": "1",
"entry": "waz up",
"username": "Bruce Willis",
"email": "bwillis@email.com",
"footer": "\n**From Microsoft Teams**",
}
def test_member_added_handler(mocker, requests_mock):
from MicrosoftTeams import member_added_handler
mocker.patch.object(demisto, "getIntegrationContext", return_value={})
mocker.patch.object(demisto, "setIntegrationContext")
requests_mock.get(f"{service_url}/v3/conversations/{team_id}/members", json=team_members)
request_body: dict = {"recipient": {"id": f"28:{bot_id}", "name": "DemistoBot"}, "membersAdded": [{"id": f"28:{bot_id}"}]}
channel_data: dict = {
"team": {"id": team_id, "name": team_name, "aadGroupId": team_aad_id},
"eventType": "teamMemberAdded",
"tenant": {"id": tenant_id},
}
member_added_handler(integration_context, request_body, channel_data)
expected_integration_context: dict = {
"bot_name": "DemistoBot",
"teams": json.dumps(
[
{
"mirrored_channels": mirrored_channels,
"team_id": team_id,
"team_aad_id": team_aad_id,
"team_members": team_members,
"team_name": team_name,
}
]
),
"tenant_id": tenant_id,
"service_url": service_url,
}
assert demisto.setIntegrationContext.call_count == 2
set_integration_context = demisto.setIntegrationContext.call_args[0]
assert len(set_integration_context) == 1
assert set_integration_context[0] == expected_integration_context
def test_mirror_investigation(mocker, requests_mock):
from MicrosoftTeams import mirror_investigation
mocker.patch.object(demisto, "results")
mocker.patch.object(demisto, "setIntegrationContext")
mocker.patch.object(demisto, "params", return_value={"team": "The-A-Team"})
# verify command cannot be executed in the war room
mocker.patch.object(demisto, "investigation", return_value={"type": 9})
with pytest.raises(ValueError) as e:
mirror_investigation()
assert str(e.value) == "Can not perform this action in playground."
# verify channel is mirrored successfully and a message is sent to it
mocker.patch.object(demisto, "investigation", return_value={"id": "2"})
channel_id: str = "channel-id"
# create channel mock request
requests_mock.post(f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels", json={"id": channel_id})
# send message mock request
requests_mock.post(f"{service_url}/v3/conversations/{channel_id}/activities", json={})
mirror_investigation()
updated_mirrored_channels: list = mirrored_channels[:]
updated_mirrored_channels.append(
{
"channel_id": "channel-id",
"investigation_id": "2",
"mirror_type": "all",
"mirror_direction": "both",
"auto_close": "true",
"mirrored": False,
"channel_name": "incident-2",
}
)
expected_integration_context: dict = {
"bot_name": "DemistoBot",
"tenant_id": tenant_id,
"service_url": service_url,
"teams": json.dumps(
[
{
"mirrored_channels": updated_mirrored_channels,
"team_id": team_id,
"team_aad_id": team_aad_id,
"team_members": team_members,
"team_name": "The-A-Team",
}
]
),
}
assert requests_mock.request_history[1].json() == {
"displayName": "incident-2",
"description": "Channel to mirror incident 2",
"membershipType": "standard",
}
assert requests_mock.request_history[3].json() == {
"text": "This channel was created to mirror [incident 2](https://test-address:8443/#/WarRoom/2) between "
"Teams and Demisto. In order for your Teams messages to be mirrored in Demisto, you need to"
" mention the Demisto Bot in the message.",
"type": "message",
}
assert demisto.setIntegrationContext.call_count == 3
set_integration_context = demisto.setIntegrationContext.call_args[0]
assert len(set_integration_context) == 1
set_integration_context[0].pop(CREDENTIALS_TOKEN_PARAMS)
set_integration_context[0].pop("bot_access_token")
set_integration_context[0].pop("bot_valid_until")
for key, value in expected_integration_context.items():
assert set_integration_context[0].get(key) == value
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == "Investigation mirrored successfully in channel incident-2."
# verify channel mirror is updated successfully
mocker.patch.object(demisto, "setIntegrationContext")
mocker.patch.object(demisto, "args", return_value={"mirror_type": "chat", "direction": "FromDemisto", "autoclose": "false"})
mocker.patch.object(demisto, "investigation", return_value={"id": "1"})
mirror_investigation()
assert demisto.setIntegrationContext.call_count == 1
set_integration_context = demisto.setIntegrationContext.call_args[0]
assert len(set_integration_context) == 1
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == "Investigation mirror was updated successfully."
# verify channel with custom channel name is mirrored successfully
mocker.patch.object(demisto, "investigation", return_value={"id": "14"})
mocker.patch.object(demisto, "args", return_value={"channel_name": "booya"})
mirror_investigation()
assert requests_mock.request_history[5].json() == {
"displayName": "booya",
"description": "Channel to mirror incident 14",
"membershipType": "standard",
}
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == "Investigation mirrored successfully in channel booya."
@pytest.mark.parametrize(
"args",
[
({"messageType": "mirrorEntry", "originalMessage": "a mirrored message\n**From Microsoft Teams**"}),
({"messageType": "incidentOpened", "severity": 1}),
],
)
def test_send_message_with_mirrored_message_or_low_severity(mocker, args):
# verify that a mirrored message is skipped
# verify notification from server with severity below threshold is not sent
from MicrosoftTeams import send_message
mocker.patch.object(demisto, "params", return_value={"min_incident_severity": "Medium", "team": "The-A-Team"})
mocker.patch.object(demisto, "args", return_value=args)
assert send_message() is None
# def test_send_message_with_low_severity(mocker):
# # verify notification from server with severity below threshold is not sent
# mocker.patch.object(
# demisto,
# 'params',
# return_value={
# 'min_incident_severity': 'Medium',
# 'team': 'The-A-Team'
# }
# )
# mocker.patch.object(
# demisto,
# 'args',
# return_value={
# 'messageType': 'incidentOpened',
# 'severity': 1
# }
# )
# assert send_message() is None
@pytest.mark.parametrize(
"args, result",
[
({}, "No channel or team member to send message were provided."),
(
{"channel": "somechannel", "team_member": "someuser"},
"Provide either channel or team member to send message to, not both.",
),
({"channel": "channel", "adaptive_card": "THISisSTRINGnotJSON"}, "Given adaptive card is not in valid JSON format."),
(
{"channel": "channel", "message": "message", "adaptive_card": '{"a":"b"}'},
"Provide either message or adaptive to send, not both.",
),
({"channel": "channel"}, "No message or adaptive card to send were provided."),
],
)
def test_send_message_raising_errors(mocker, args, result):
# verify error is raised if no user/channel were provided.
# verify error is raised if user and channel provided.
# verify proper error is raised if invalid JSON provided as adaptive card.
# verify proper error is raised if both message and adaptive card were provided.
# verify proper error is raised if neither message or adaptive card were provided.
from MicrosoftTeams import send_message
mocker.patch.object(demisto, "args", return_value=args)
with pytest.raises(ValueError) as e:
send_message()
assert result in str(e.value)
@pytest.mark.parametrize(
"message", ["MESSAGE", "891f1e9d-b8c3-4e24-bfbb-c44bcca4d586", "testing 891f1e9d-b8c3-4e24-bfbb-c44bcca4d586 testing"]
)
def test_send_message_with_user(mocker, requests_mock, message):
"""
Given:
- a message as a basic string and a message that contains GUID.
When:
- running send message function.
Then:
- The message is sent successfully in both cases.
"""
# verify message is sent properly given user to send to
from MicrosoftTeams import send_message
mocker.patch.object(demisto, "results")
expected = util_load_json("test_data/send_message/expected_generic.json")
raw = util_load_json("test_data/send_message/raw_generic.json")
mocker.patch("MicrosoftTeams.BOT_ID", new=bot_id)
mocker.patch.object(demisto, "args", return_value={"team_member": "Denzel Washington", "message": message})
requests_mock.post(f"{service_url}/v3/conversations", json={"id": "conversation-id"})
requests_mock.post(f"{service_url}/v3/conversations/conversation-id/activities", json=raw)
expected_create_personal_conversation_data: dict = {
"bot": {"id": f"28:{bot_id}", "name": "DemistoBot"},
"members": [{"id": "29:1pBMMC85IyjM3tr_MCZi7KW4pw4EULxLN4C7R_xoi3Wva_lOn3VTf7xJlCLK-r-pMumrmoz9agZxsSrCf7__u9R"}],
"channelData": {"tenant": {"id": tenant_id}},
}
send_message()
assert requests_mock.request_history[0].json() == expected_create_personal_conversation_data
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == expected
def test_send_message_with_channel(mocker, requests_mock):
# verify message is sent properly given channel
from MicrosoftTeams import send_message
mocker.patch.object(demisto, "results")
mocker.patch("MicrosoftTeams.get_channel_type", return_value="standard")
expected = util_load_json("test_data/send_message/expected_generic.json")
raw = util_load_json("test_data/send_message/raw_generic.json")
mocker.patch.object(demisto, "params", return_value={"team": "The-A-Team"})
mocker.patch.object(demisto, "args", return_value={"channel": "incident-1", "message": "MESSAGE"})
requests_mock.post(f"{service_url}/v3/conversations/{mirrored_channels[0]['channel_id']}/activities", json=raw)
send_message()
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == expected
def test_send_message_with_entitlement(mocker, requests_mock):
# verify message is sent properly given entitlement
from MicrosoftTeams import send_message
mocker.patch.object(demisto, "results")
expected = util_load_json("test_data/send_message/expected_generic.json")
raw = util_load_json("test_data/send_message/raw_generic.json")
message: dict = {
"message_text": "is this really working?",
"options": ["yes", "no", "maybe"],
"entitlement": "4404dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "72",
"task_id": "23",
"form_type": "predefined-options",
}
mocker.patch.object(demisto, "args", return_value={"team_member": "dwashinton@email.com", "message": json.dumps(message)})
requests_mock.post(f"{service_url}/v3/conversations", json={"id": "conversation-id"})
requests_mock.post(f"{service_url}/v3/conversations/conversation-id/activities", json=raw)
expected_ask_user_message: dict = {
"attachments": [
{
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"actions": [
{
"data": {
"entitlement": "4404dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "72",
"response": "yes",
"task_id": "23",
},
"title": "yes",
"type": "Action.Submit",
},
{
"data": {
"entitlement": "4404dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "72",
"response": "no",
"task_id": "23",
},
"title": "no",
"type": "Action.Submit",
},
{
"data": {
"entitlement": "4404dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "72",
"response": "maybe",
"task_id": "23",
},
"title": "maybe",
"type": "Action.Submit",
},
],
"body": [{"text": "is this really working?", "type": "TextBlock", "wrap": True}],
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"version": "1.0",
},
"contentType": "application/vnd.microsoft.card.adaptive",
}
],
"type": "message",
}
send_message()
assert requests_mock.request_history[1].json() == expected_ask_user_message
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == expected
def test_send_message_with_adaptive_card(mocker, requests_mock):
# verify adaptive card sent successfully
from MicrosoftTeams import send_message
mocker.patch.object(demisto, "results")
expected = util_load_json("test_data/send_message/expected_generic.json")
raw = util_load_json("test_data/send_message/raw_generic.json")
adaptive_card: dict = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"type": "AdaptiveCard",
"version": "1.0",
"body": [{"type": "Container", "items": [{"type": "TextBlock", "text": "What a pretty adaptive card"}]}],
},
}
mocker.patch.object(
demisto, "args", return_value={"team_member": "bwillis@email.com", "adaptive_card": json.dumps(adaptive_card)}
)
requests_mock.post(f"{service_url}/v3/conversations", json={"id": "conversation-id"})
requests_mock.post(f"{service_url}/v3/conversations/conversation-id/activities", json=raw)
send_message()
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == expected
def test_sending_message_using_email_address(mocker, requests_mock):
from MicrosoftTeams import send_message
mocker.patch.object(demisto, "results")
expected = util_load_json("test_data/send_message/expected_generic.json")
raw = util_load_json("test_data/send_message/raw_generic.json")
# verify message is sent properly given email with uppercase letters to send to
mocker.patch("MicrosoftTeams.BOT_ID", new=bot_id)
mocker.patch.object(demisto, "args", return_value={"team_member": "DwashinTon@email.com", "message": "MESSAGE"})
requests_mock.post(f"{service_url}/v3/conversations", json={"id": "conversation-id"})
requests_mock.post(f"{service_url}/v3/conversations/conversation-id/activities", json=raw)
expected_create_personal_conversation_data: dict = {
"bot": {"id": f"28:{bot_id}", "name": "DemistoBot"},
"members": [{"id": "29:1pBMMC85IyjM3tr_MCZi7KW4pw4EULxLN4C7R_xoi3Wva_lOn3VTf7xJlCLK-r-pMumrmoz9agZxsSrCf7__u9R"}],
"channelData": {"tenant": {"id": tenant_id}},
}
send_message()
assert requests_mock.request_history[0].json() == expected_create_personal_conversation_data
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == expected
def test_send_message_server_notifications_incident_opened(mocker, requests_mock):
"""
Given:
- Notification from server of an incident opened.
When:
- Sending notification message of the incident opened.
Then:
- Ensure message is sent successfully.
- Verify the message is sent to the dedicated notifications channel.
"""
from MicrosoftTeams import send_message
mocker.patch.object(demisto, "results")
mocker.patch("MicrosoftTeams.get_channel_type", return_value="standard")
expected = util_load_json("test_data/send_message/expected_generic.json")
raw = util_load_json("test_data/send_message/raw_generic.json")
mocker.patch.object(
demisto,
"params",
return_value={"team": "The-A-Team", "min_incident_severity": "Low", "incident_notifications_channel": "General"},
)
mocker.patch.object(
demisto,
"args",
return_value={
"channel": "incidentNotificationChannel",
"message": "user has reported an incident tadam.\nView it on https://server/#/WarRoom/3247",
"messageType": "incidentOpened",
"severity": 1,
"to": "",
},
)
requests_mock.get(
f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels",
json={
"value": [
{
"description": "general channel",
"displayName": "General",
"id": "19:67pd3966e74g45f28d0c65f1689132bb@thread.skype",
}
]
},
)
requests_mock.post(f"{service_url}/v3/conversations/19:67pd3966e74g45f28d0c65f1689132bb@thread.skype/activities", json=raw)
send_message()
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == expected
def test_send_message_server_notifications_incident_changed(mocker, requests_mock):
"""
Given:
- Notification from server of an updated incident.
When:
- Sending notification message of the updated incident.
Then:
- Ensure message is sent successfully.
- Verify the message is sent to the dedicated notifications channel.
"""
from MicrosoftTeams import send_message
mocker.patch.object(demisto, "results")
mocker.patch("MicrosoftTeams.get_channel_type", return_value="standard")
expected = util_load_json("test_data/send_message/expected_generic.json")
raw = util_load_json("test_data/send_message/raw_generic.json")
mocker.patch.object(
demisto,
"params",
return_value={"team": "The-A-Team", "min_incident_severity": "Low", "incident_notifications_channel": "General"},
)
mocker.patch.object(
demisto,
"args",
return_value={
"channel": "incidentNotificationChannel",
"message": "DBot has updated an incident tadam.\nView it on https://server/#/WarRoom/3247",
"messageType": "incidentChanged",
"severity": 1,
"to": "",
},
)
requests_mock.get(
f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels",
json={
"value": [
{
"description": "general channel",
"displayName": "General",
"id": "19:67pd3966e74g45f28d0c65f1689132bb@thread.skype",
}
]
},
)
requests_mock.post(f"{service_url}/v3/conversations/19:67pd3966e74g45f28d0c65f1689132bb@thread.skype/activities", json=raw)
send_message()
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == expected
def test_get_channel_id(requests_mock):
from MicrosoftTeams import get_channel_id
# get channel which is in the integration context
assert get_channel_id("incident-1", team_aad_id) == "19:2cbad0d78c624400ef83a5750539998g@thread.skype"
# get channel which is not in the integration context
requests_mock.get(
f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels",
json={
"value": [
{
"description": "channel for incident 1",
"displayName": "incident-1",
"id": "19:67pd3967e74g45f28d0c65f1689132bb@thread.skype",
},
{
"description": "channel for incident 2",
"displayName": "incident-3",
"id": "19:67pd3967e74g45f28d0c65f1689132bo@thread.skype",
},
]
},
)
assert get_channel_id("incident-3", team_aad_id) == "19:67pd3967e74g45f28d0c65f1689132bo@thread.skype"
# Try a channel which does not exit
with pytest.raises(ValueError) as e:
get_channel_id("incident-4", team_aad_id)
assert str(e.value) == "Could not find channel: incident-4"
def test_close_channel(mocker, requests_mock):
from MicrosoftTeams import close_channel
requests_mock.delete(
f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels/19:2cbad0d78c624400ef83a5750539998g@thread.skype",
status_code=204,
)
requests_mock.delete(
f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels/19:2cbad0d78c624400ef83a5750534448g@thread.skype",
status_code=204,
)
mocker.patch.object(demisto, "results")
# close channel without given channel name
mocker.patch.object(demisto, "investigation", return_value={"id": "1"})
mocker.patch.object(demisto, "getIntegrationContext", return_value=integration_context)
mocker.patch.object(demisto, "setIntegrationContext")
close_channel()
assert requests_mock.request_history[0].method == "DELETE"
assert demisto.setIntegrationContext.call_count == 1
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == "Channel was successfully closed."
# try to close channel without given channel name, which does not exist in the integration context
mocker.patch.object(demisto, "investigation", return_value={"id": "5"})
with pytest.raises(ValueError) as e:
close_channel()
assert str(e.value) == "Could not find Microsoft Teams channel to close."
# close channel given channel name
mocker.patch.object(demisto, "results")
mocker.patch.object(demisto, "setIntegrationContext")
requests_mock.get(
f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels",
json={
"value": [
{
"description": "channel for incident 1",
"displayName": "incident-1",
"id": "19:67pd3967e74g45f28d0c65f1689132bb@thread.skype",
},
{
"description": "channel for incident 6",
"displayName": "incident-6",
"id": "19:67pd3967e74g45f28d0c65f1689132bo@thread.skype",
},
]
},
)
requests_mock.delete(
f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels/19:67pd3967e74g45f28d0c65f1689132bb@thread.skype",
status_code=204,
)
mocker.patch.object(demisto, "params", return_value={"team": "The-A-Team"})
mocker.patch.object(demisto, "args", return_value={"channel": "incident-1"})
close_channel()
assert requests_mock.request_history[0].method == "DELETE"
assert demisto.setIntegrationContext.call_count == 0
assert demisto.results.call_count == 1
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == "Channel was successfully closed."
def test_entitlement_handler(mocker, requests_mock):
from MicrosoftTeams import entitlement_handler
mocker.patch.object(demisto, "handleEntitlementForUser")
conversation_id: str = "f:3005393407786078157"
activity_id: str = "1:1vW2mx4iDZf05lk18yskL64Wkfwraa76YTGNgDiIi-_5"
requests_mock.put(f"{service_url}/v3/conversations/{conversation_id}/activities/{activity_id}", json={"id": "updateid"})
request_body: dict = {
"from": {
"id": "29:1KZccCJRTxlPdHnwcKfxHAtYvPLIyHgkSLhFSnGXLGVFlnltovdZPmZAduPKQP6NrGqOcde7FXAF7uTZ_8FQOqg",
"aadObjectId": "359d2c3c-162b-414c-b2eq-386461e5l050",
"name": "Bruce Willis",
},
"replyToId": activity_id,
}
value: dict = {
"response": "Approve!",
"entitlement": "4404dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "100",
"task_id": "4",
}
entitlement_handler(integration_context, request_body, value, conversation_id)
assert demisto.handleEntitlementForUser.call_count == 1
handle_entitlement_args = demisto.handleEntitlementForUser.call_args[1]
assert handle_entitlement_args == {
"incidentID": "100",
"guid": "4404dae8-2d45-46bd-85fa-64779c12abe8",
"taskID": "4",
"email": "bwillis@email.com",
"content": "Approve!",
}
def test_translate_severity():
from MicrosoftTeams import translate_severity
assert translate_severity("Low") == 1
assert translate_severity("NotRealSeverity") == 0
def test_is_investigation_mirrored():
from MicrosoftTeams import is_investigation_mirrored
existing_investigation_id: str = "1"
non_existing_investigation_id: str = "2"
assert is_investigation_mirrored(existing_investigation_id, mirrored_channels) == 0
assert is_investigation_mirrored(non_existing_investigation_id, mirrored_channels) == -1
@pytest.mark.parametrize(
"message, expected_result",
[
(
"Visit https://github.com/demisto/content and https://xsoar.pan.dev",
"Visit [https://github.com/demisto/content](https://github.com/demisto/content) and "
"[https://xsoar.pan.dev](https://xsoar.pan.dev)",
),
(
"Link: https://xsoar.pan.dev/page?parametized=true",
"Link: [https://xsoar.pan.dev/page?parametized=true](https://xsoar.pan.dev/page?parametized=true)",
),
(
"This is a link https://paloaltonetworks.com/. This is a [Custom URL](https://paloaltonetworks.com/)",
"This is a link [https://paloaltonetworks.com/.](https://paloaltonetworks.com/.) This is a [Custom URL]("
"https://paloaltonetworks.com/)",
),
(
"This is a [Custom URL](https://paloaltonetworks.com/), This is a link https://paloaltonetworks.com/",
"This is a [Custom URL](https://paloaltonetworks.com/), "
"This is a link [https://paloaltonetworks.com/](https://paloaltonetworks.com/)",
),
],
)
def test_urlify_hyperlinks(message: str, expected_result: str):
from MicrosoftTeams import urlify_hyperlinks
assert urlify_hyperlinks(message) == expected_result
def test_get_team_aad_id(mocker, requests_mock):
from MicrosoftTeams import get_team_aad_id
# verify team ID for team which is in integration context
mocker.patch.object(demisto, "params", return_value={"team": "The-A-Team"})
assert get_team_aad_id("The-A-Team") == "7d8efdf8-0c5a-42e3-a489-5ef5c3fc7a2b"
json_response = {
"@odata.context": "https://graph.microsoft.com/v1.0/$metadata#groups",
"value": [
{
"id": "02bd9fd6-8f93-4758-87c3-1fb73740a315",
"displayName": "MyGreat #Team",
"groupTypes": ["Unified"],
"mailEnabled": True,
"resourceBehaviorOptions": [],
"resourceProvisioningOptions": ["Team"],
"securityEnabled": False,
"visibility": "Private",
},
{
"id": "8090c93e-ba7c-433e-9f39-08c7ba07c0b3",
"displayName": "WooahTeam",
"groupTypes": ["Unified"],
"mailEnabled": True,
"mailNickname": "X1050LaunchTeam",
"resourceBehaviorOptions": [],
"resourceProvisioningOptions": ["Team"],
"securityEnabled": False,
"visibility": "Private",
},
],
}
# verify non existing team raises value error
url_a = f"{BASE_URL}?$filter=displayName eq 'The-B-Team' and resourceProvisioningOptions/Any(x:x eq 'Team')"
requests_mock.get(url_a, json=json_response)
with pytest.raises(ValueError) as e:
get_team_aad_id("The-B-Team")
assert str(e.value) == "Could not find requested team."
url_b = f"{BASE_URL}?$filter=displayName eq 'MyGreat%20%23Team' and resourceProvisioningOptions/Any(x:x eq 'Team')"
requests_mock.get(url_b, json=json_response)
# verify team ID for team which is not in integration context
assert get_team_aad_id("MyGreat #Team") == "02bd9fd6-8f93-4758-87c3-1fb73740a315"
def test_get_team_member():
from MicrosoftTeams import get_team_member
user_id: str = "29:1KZccCJRTxlPdHnwcKfxHAtYvPLIyHgkSLhFSnGXLGVFlnltovdZPmZAduPKQP6NrGqOcde7FXAF7uTZ_8FQOqg"
team_member: dict = {
"username": "Bruce Willis",
"user_email": "bwillis@email.com",
"user_principal_name": "bwillis@email.com",
}
assert get_team_member(integration_context, user_id) == team_member
with pytest.raises(ValueError) as e:
get_team_member(integration_context, "NotRealUser")
assert str(e.value) == "Team member was not found"
def test_get_team_member_id():
from MicrosoftTeams import get_team_member_id
requested_team_member: str = "Denzel Washington"
expected_user_id: str = "29:1pBMMC85IyjM3tr_MCZi7KW4pw4EULxLN4C7R_xoi3Wva_lOn3VTf7xJlCLK-r-pMumrmoz9agZxsSrCf7__u9R"
assert get_team_member_id(requested_team_member, integration_context) == expected_user_id
requested_team_member = "dwashinton@email.com"
assert get_team_member_id(requested_team_member, integration_context) == expected_user_id
requested_team_member = "TheRock"
with pytest.raises(ValueError) as e:
get_team_member_id(requested_team_member, integration_context)
assert str(e.value) == "Team member TheRock was not found"
def test_create_adaptive_card():
from MicrosoftTeams import create_adaptive_card
body: list = [{"type": "TextBlock", "size": "Medium", "weight": "Bolder", "text": "What a beautiful text"}]
expected_adaptive_card: dict = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": body,
},
}
assert create_adaptive_card(body) == expected_adaptive_card
actions: list = [{"type": "Action.OpenUrl", "title": "DEMISTO", "url": "https://www.demisto.com"}]
expected_adaptive_card["content"]["actions"] = actions
assert create_adaptive_card(body, actions) == expected_adaptive_card
def test_process_tasks_list():
from MicrosoftTeams import process_tasks_list
data_by_line: list = [
"Task | Incident | Due | Link ",
"=========================================|================================|=====================|=====",
"Manually review the incident | 21 - nnn | 0001-01-01 00:00:00 | "
"https://demisto.com/#/WorkPlan/21",
]
expected_adaptive_card: dict = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": [
{
"type": "FactSet",
"facts": [
{"title": "Task:", "value": "Manually review the incident"},
{"title": "Incident:", "value": "21 - nnn"},
{"title": "Due:", "value": "0001-01-01 00:00:00"},
{"title": "Link:", "value": "[https://demisto.com/#/WorkPlan/21](https://demisto.com/#/WorkPlan/21)"},
],
}
],
},
}
assert process_tasks_list(data_by_line) == expected_adaptive_card
def test_process_incidents_list():
from MicrosoftTeams import process_incidents_list
data_by_line: list = [
"ID | Name | Status | Type | Owner | Created | Link ",
"===========|======================|=============|=============|=============|=====================|=====",
"257 | w | Active | Unclassifie | god | 2019-07-28 16:42:40 | "
"https://demisto.com/#/WarRoom/257",
"250 | gosa | Active | Unclassifie | mozes | 2019-07-28 16:16:49 | "
"https://demisto.com/#/WarRoom/250 ",
]
expected_adaptive_card: dict = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": [
{
"type": "FactSet",
"facts": [
{"title": "ID:", "value": "257"},
{"title": "Name:", "value": "w"},
{"title": "Status:", "value": "Active"},
{"title": "Type:", "value": "Unclassifie"},
{"title": "Owner:", "value": "god"},
{"title": "Created:", "value": "2019-07-28 16:42:40"},
{"title": "Link:", "value": "[https://demisto.com/#/WarRoom/257](https://demisto.com/#/WarRoom/257)"},
],
},
{
"type": "FactSet",
"facts": [
{"title": "ID:", "value": "250"},
{"title": "Name:", "value": "gosa"},
{"title": "Status:", "value": "Active"},
{"title": "Type:", "value": "Unclassifie"},
{"title": "Owner:", "value": "mozes"},
{"title": "Created:", "value": "2019-07-28 16:16:49"},
{"title": "Link:", "value": "[https://demisto.com/#/WarRoom/250](https://demisto.com/#/WarRoom/250)"},
],
},
],
},
}
assert process_incidents_list(data_by_line) == expected_adaptive_card
def test_process_mirror_or_unknown_message():
from MicrosoftTeams import process_mirror_or_unknown_message
message: str = (
"I can understand the following commands:\nlist incidents [page x]\nlist my incidents [page x]\n"
"list my tasks\nlist closed incidents\nnew incident [details]\nmirror incident-id"
)
expected_adaptive_card: dict = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": [
{
"type": "TextBlock",
"text": "I can understand the following commands:\n\nlist incidents [page x]\n\nlist my incidents [page"
" x]\n\nlist my tasks\n\nlist closed incidents\n\nnew incident [details]\n\nmirror incident-id",
"wrap": True,
}
],
},
}
assert process_mirror_or_unknown_message(message) == expected_adaptive_card
def test_get_participant_info():
from MicrosoftTeams import get_participant_info
participants = {
"organizer": {
"upn": "mail.com",
"role": "presenter",
"identity": {
"phone": None,
"guest": None,
"encrypted": None,
"onPremises": None,
"applicationInstance": None,
"application": None,
"device": None,
"user": {
"id": "id_identifier",
"displayName": "best_user",
"tenantId": "tenantId_identifier",
"identityProvider": "AAD",
},
},
},
"attendees": [],
}
participant_id, participant_display_name = get_participant_info(participants)
assert participant_id == "id_identifier"
assert participant_display_name == "best_user"
def test_create_channel(requests_mock):
from MicrosoftTeams import create_channel
requests_mock.post(
f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels",
json={"id": "19:67pd3967e74g45f28d0c65f1689132bb@thread.skype"},
)
channel_name: str = "CrazyChannel"
response = create_channel(team_aad_id, channel_name)
assert response == "19:67pd3967e74g45f28d0c65f1689132bb@thread.skype"
def test_create_meeting_command(requests_mock, mocker):
from MicrosoftTeams import create_meeting_command
mocker.patch.object(demisto, "args", return_value={"subject": "Best_Meeting", "member": "username"})
mocker.patch.object(demisto, "results")
requests_mock.get("https://graph.microsoft.com/v1.0/users", json={"value": [{"id": "userid1"}]})
requests_mock.post(
"https://graph.microsoft.com/v1.0/users/userid1/onlineMeetings",
json={
"chatInfo": {"threadId": "19:@thread.skype", "messageId": "0", "replyChainMessageId": "0"},
"creationDateTime": "2019-07-11T02:17:17.6491364Z",
"startDateTime": "2019-07-11T02:17:17.6491364Z",
"endDateTime": "2019-07-11T02:47:17.651138Z",
"id": "id_12345",
"joinWebUrl": "https://teams.microsoft.com/l/meetup-join/12345",
"participants": {
"organizer": {"identity": {"user": {"id": "user_id_12345", "displayName": "Demisto"}}, "upn": "upn-value"}
},
"subject": "User Token Meeting",
},
)
expected_results = 'The meeting "Best_Meeting" was created successfully'
create_meeting_command()
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0]["HumanReadable"] == expected_results
assert results[0]["Contents"].get("id") == "id_12345"
def test_get_team_members(requests_mock):
from MicrosoftTeams import get_team_members
requests_mock.get(f"{service_url}/v3/conversations/{team_aad_id}/members", json=team_members)
assert get_team_members(service_url, team_aad_id) == team_members
def test_update_message(requests_mock):
from MicrosoftTeams import update_message
activity_id: str = "1:1vW2mx4iDZf05lk18yskL64Wkfwraa76YTGNgDiIi-_5"
conversation_id: str = "f:3005393407786078157"
requests_mock.put(f"{service_url}/v3/conversations/{conversation_id}/activities/{activity_id}", json={"id": "updateid"})
expected_conversation: dict = {
"type": "message",
"attachments": [
{
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": [{"type": "TextBlock", "text": "OMG!"}],
},
}
],
}
update_message(service_url, conversation_id, activity_id, "OMG!")
assert requests_mock.request_history[0].method == "PUT"
assert json.loads(requests_mock.request_history[0].body) == expected_conversation
# def test_create_team(mocker, requests_mock):
# from MicrosoftTeams import create_team
# mocker.patch.object(
# demisto,
# 'args',
# return_value={
# 'display_name': 'OhMyTeam',
# 'mail_nickname': 'NoNicknamesPlease',
# 'owner': 'nonexistingmmember@demisto.com',
# 'mail_enabled': 'true',
# 'security_enabled': 'false'
# }
# )
# requests_mock.get(
# f'https://graph.microsoft.com/v1.0/users',
# json={
# 'value': team_members
# }
# )
# with pytest.raises(ValueError) as e:
# create_team()
# assert str(e.value) == 'Could not find given users to be Team owners.'
# mocker.patch.object(
# demisto,
# 'args',
# return_value={
# 'display_name': 'OhMyTeam',
# 'mail_nickname': 'NoNicknamesPlease',
# 'owner': 'dwashinton@email.com'
# }
# )
def test_direct_message_handler(mocker, requests_mock):
from MicrosoftTeams import direct_message_handler
mocker.patch("MicrosoftTeams.get_graph_access_token", return_value="token")
mocker.patch("MicrosoftTeams.get_bot_access_token", return_value="token")
create_incidents_mocker = mocker.patch.object(
demisto, "createIncidents", return_value={"id": "4", "name": "incidentnumberfour"}
)
requests_mock.post(f"{service_url}/v3/conversations/conversation-id/activities", json={})
request_body: dict = {
"from": {"id": "29:1KZccCJRTxlPdHnwcKfxHAtYvPLIyHgkSLhFSnGXLGVFlnltovdZPmZAduPKQP6NrGqOcde7FXAF7uTZ_8FQOqg"}
}
conversation: dict = {"id": "conversation-id"}
expected_created_incident: list = [
{
"name": "GoFish",
"type": "Phishing",
"rawJSON": '{"from": {"id": '
'"29:1KZccCJRTxlPdHnwcKfxHAtYvPLIyHgkSLhFSnGXLGVFlnltovdZPmZAduPKQP6NrGqOcde7FXAF7uTZ_8FQOqg", '
'"username": "Bruce Willis", "user_email": "bwillis@email.com", '
'"user_principal_name": "bwillis@email.com"}}',
}
]
expected_assigned_user = "nice-demisto-id"
# verify create incident fails on un allowed external incident creation and non found user
message: str = "create incident name=GoFish type=Phishing"
mocker.patch.object(demisto, "findUser", return_value=None)
direct_message_handler(integration_context, request_body, conversation, message)
response = requests_mock.request_history[0].json()
assert response["type"] == "message"
assert (
response["text"] == "I'm sorry but I was unable to find you as a Cortex XSOAR user for bwillis@email.com. "
"You're not allowed to run any command"
)
# verify create incident successfully
mocker.patch.object(demisto, "findUser", return_value={"id": "nice-demisto-id"})
direct_message_handler(integration_context, request_body, conversation, message)
response = requests_mock.request_history[1].json()
assert response["type"] == "message"
assert (
response["text"] == "Successfully created incident incidentnumberfour.\n"
"View it on: [https://test-address:8443/#/WarRoom/4]"
"(https://test-address:8443/#/WarRoom/4)"
)
create_incidents_mocker.assert_called_with(expected_created_incident, userID=expected_assigned_user)
# verify get my incidents
my_incidents: str = (
"```ID | Name | Status | Type | Owner | Created"
" | Link\n ===========|======================|=============|=============|===="
"=========|=====================|=====\n257 | w | Active | "
"Unclassifie | god | 2019-07-28 16:42:40 | https://demisto.com/#/WarRoom/257```"
)
mocker.patch.object(demisto, "directMessage", return_value=my_incidents)
message = "list my incidents"
direct_message_handler(integration_context, request_body, conversation, message)
assert requests_mock.request_history[2].json() == {
"attachments": [
{
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"body": [
{
"facts": [
{"title": "ID:", "value": "257"},
{"title": "Name:", "value": "w"},
{"title": "Status:", "value": "Active"},
{"title": "Type:", "value": "Unclassifie"},
{"title": "Owner:", "value": "god"},
{"title": "Created:", "value": "2019-07-28 16:42:40"},
{
"title": "Link:",
"value": "[https://demisto.com/#/WarRoom/257](https://demisto.com/#/WarRoom/257)",
},
],
"type": "FactSet",
}
],
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"version": "1.0",
},
"contentType": "application/vnd.microsoft.card.adaptive",
}
],
"type": "message",
}
# verify error message raised by Demisto server is sent as message as expected
mocker.patch.object(
demisto,
"directMessage",
side_effect=ValueError("I'm sorry but I was unable to find you as a Demisto user for email [johnnydepp@gmail.com]"),
)
direct_message_handler(integration_context, request_body, conversation, message)
assert requests_mock.request_history[3].json() == {
"type": "message",
"text": "I'm sorry but I was unable to find you as a Demisto user for email [johnnydepp@gmail.com]",
}
def test_error_parser():
from MicrosoftTeams import error_parser
class MockResponse:
def __init__(self, json_data, status_code, text=""):
self.json_data = json_data
self.status_code = status_code
self.text = text
def json(self):
return self.json_data
# verify bot framework error parsed as expected
error_description: str = (
"AADSTS700016: Application with identifier '2bc5202b-ad6a-4458-8821-e104af433bbb' "
"was not found in the directory 'botframework.com'. This can happen if the application "
"has not been installed by the administrator of the tenant or consented to by any user "
"in the tenant. You may have sent your authentication request to the wrong tenant.\r\n"
"Trace ID: 9eaeeec8-7f9e-4fb8-b319-5413581f0a00\r\nCorrelation ID: "
"138cb511-2484-410e-b9c1-14b15accbeba\r\nTimestamp: 2019-08-28 13:18:44Z"
)
bot_error_json_response: dict = {
"error": "unauthorized_client",
"error_description": error_description,
"error_codes": [700016],
"timestamp": "2019-08-28 13:18:44Z",
"trace_id": "9eaeeec8-7f9e-4fb8-b319-5413581f0a11",
"correlation_id": "138cb111-2484-410e-b9c1-14b15accbeba",
"error_uri": "https://login.microsoftonline.com/error?code=700016",
}
bot_error_json_response = MockResponse(bot_error_json_response, 400)
assert error_parser(bot_error_json_response, "bot") == error_description
# verify graph error parsed as expected
error_code: str = "InvalidAuthenticationToken"
error_message: str = "Access token validation failure."
graph_error_json_response: dict = {
"error": {
"code": error_code,
"message": error_message,
"innerError": {"request-id": "c240ab22-4463-4a1f-82bc-8509d8190a77", "date": "2019-08-28T13:37:14"},
}
}
graph_error_json_response = MockResponse(graph_error_json_response, 401)
assert error_parser(graph_error_json_response) == f"{error_code}: {error_message}"
def test_integration_health(mocker):
from MicrosoftTeams import integration_health
mocker.patch.object(demisto, "results")
expected_results = """### Microsoft API Health
|Bot Framework API Health|Graph API Health|
|---|---|
| Operational | Operational |
### Microsoft Teams Mirrored Channels
|Channel|Investigation ID|Team|
|---|---|---|
| incident-10 | 10 | The-A-Team |
| incident-2 | 2 | The-A-Team |
| booya | 14 | The-A-Team |
"""
integration_health()
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0]["HumanReadable"] == expected_results
def load_test_data(path):
with open(path) as f:
return json.load(f)
@pytest.mark.parametrize(
"args, mock_res_create_channel",
[
(
{
"channel_name": "Private Channel",
"description": "Private Channel test",
"team": "TestTeam",
"membership_type": "private",
},
{"id": "channel_id"},
),
(
{
"channel_name": "Shared Channel",
"description": "Shared Channel test",
"team": "TestTeam",
"membership_type": "shared",
},
{},
),
(
{"channel_name": "Default Standard Channel", "description": "Standard Channel test", "team": "TestTeam"},
{"id": "channel_id"},
),
(
{
"channel_name": "Standard Channel",
"description": "Standard Channel test",
"team": "TestTeam",
"membership_type": "standard",
"owner_user": "jacob@contoso.com",
},
{"id": "channel_id"},
),
],
)
def test_create_channel_command(mocker, requests_mock, args, mock_res_create_channel):
"""
Given:
- case 1: request to create private channel without specify owner_user
- case 2: request to create shared channel without specify owner_user
- case 3: request to create standard channel without specify owner_user and membership_type
- case 4: request to create standard channel with specify owner_user
When:
- Executing the 'microsoft-teams-create-channel' command.
Then:
- Ensure expected request body (in the post request to create channel) is sent.
- Verify human-readable output
"""
from MicrosoftTeams import create_channel_command
mocker.patch.object(demisto, "args", return_value=args)
owner_user = demisto.args().get("owner_user")
channel_name = demisto.args().get("channel_name")
channel_description = demisto.args().get("description")
membership_type = demisto.args().get("membership_type", "standard")
mocker.patch("MicrosoftTeams.get_user", return_value=[{"id": "user_id", "userType": "Member"}])
mocker.patch("MicrosoftTeams.get_team_aad_id", return_value=team_aad_id)
mocker.patch("MicrosoftTeams.AUTH_TYPE", new=AUTHORIZATION_CODE_FLOW)
# create_channel mock request
requests_mock.post(
f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels",
json=mock_res_create_channel, # The response object shown here is shortened.
)
expected_json = {"displayName": channel_name, "description": channel_description, "membershipType": membership_type}
if owner_user:
expected_json["members"] = [
{
"@odata.type": "#microsoft.graph.aadUserConversationMember",
"user@odata.bind": "https://graph.microsoft.com/v1.0/users('user_id')",
"roles": ["owner"],
}
]
mocker.patch.object(demisto, "results")
create_channel_command()
assert requests_mock.request_history[0].json() == expected_json
expected_results = f'The channel "{channel_name}" was created successfully'
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == expected_results
@pytest.mark.parametrize(
"args, expected_error",
[
(
{"channel_name": "test private", "team": "TeamTest", "membership_type": "private"},
"When using the 'Client Credentials flow', you must specify an 'owner_user'.",
),
(
{"channel_name": "test private", "team": "TeamTest", "membership_type": "private", "owner_user": "no_user"},
'The given owner_user "no_user" was not found',
),
],
)
def test_create_channel_command_errors(mocker, args, expected_error):
"""
Given:
- The command arguments without the 'owner_user' argument using the 'Client Credentials flow'
- The command arguments with 'owner_user' that doesn't exist
When:
- Executing the 'microsoft-teams-create-channel' command.
Then:
- The expected error is raised
"""
from MicrosoftTeams import create_channel_command
mocker.patch.object(demisto, "args", return_value=args)
mocker.patch("MicrosoftTeams.get_user", return_value=[])
with pytest.raises(ValueError) as e:
create_channel_command()
assert str(e.value) == expected_error
expected_hr_user_list = (
'### Channel "Test Channel" Members List:\n'
"|User Id|Email|Tenant Id|Membership id|User roles|Display Name|Start "
"DateTime|\n"
"|---|---|---|---|---|---|---|\n"
"| eef9cb36-06de-469b-87cd-70f4cbe32d14 | jdoe@teamsip.onmicrosoft.com | "
"pbae9ao6-01ql-249o-5me3-4738p3e1m941 | "
"MmFiOWM3OTYtMjkwMi00NWY4LWI3MTItN2M1YTYzY2Y0MWM0IyNlZWY5Y2IzNi0wNmRlLTQ2OWItODdjZC03MGY0Y2JlMzJkMTQ= "
"| | Jane Doe | 0001-01-01T00:00:00Z |\n"
"| b3246f44-c091-4627-96c6-25b18fa2c910 | ajohn@teamsip.onmicrosoft.com | "
"pbae9ao6-01ql-249o-5me3-4738p3e1m941 | "
"MmFiOWM3OTYtMjkwMi00NWY4LWI3MTItN2M1YTYzY2Y0MWM0IyNiMzI0NmY0NC1jMDkxLTQ2MjctOTZjNi0yNWIxOGZhMmM5MTA= "
"| owner | Ace John | 0001-01-01T00:00:00Z |\n"
)
def test_channel_user_list_command(mocker):
"""
Given:
- The command arguments
When:
- Executing the 'microsoft-teams-channel-user-list' command.
Then:
- Verify human-readable output
- Verify entry context output
"""
from MicrosoftTeams import channel_user_list_command
mocker.patch.object(demisto, "args", return_value={"channel_name": "Test Channel", "team": "TestTeam"})
return_results = mocker.patch("MicrosoftTeams.return_results")
mocker.patch("MicrosoftTeams.get_team_aad_id", return_value=team_aad_id)
mocker.patch("MicrosoftTeams.get_channel_id", return_value=channel_id)
get_channel_members_expected_response = channel_members.get("value", [])
mocker.patch("MicrosoftTeams.get_channel_members", return_value=get_channel_members_expected_response)
channel_user_list_command()
results_hr = return_results.call_args[0][0].readable_output
results_outputs = return_results.call_args[0][0].outputs
assert results_hr == expected_hr_user_list
[member.pop("@odata.type", None) for member in get_channel_members_expected_response]
expected_outputs = {"channelName": "Test Channel", "channelId": channel_id, "members": get_channel_members_expected_response}
assert results_outputs == expected_outputs
def test_get_channel_members(requests_mock):
"""
Given:
- The function arguments team_id, channel_id
When:
- Calling the get_channel_members function
Then:
- The function returns the expected value
"""
from MicrosoftTeams import get_channel_members
requests_mock.get(f"{GRAPH_BASE_URL}/v1.0/teams/{team_aad_id}/channels/{channel_id}/members", json=channel_members)
assert get_channel_members(team_aad_id, channel_id) == channel_members.get("value", [])
membership_id = "ZWUwZjVhZTItOGJjNi00YWU1LTg0NjYtN2RhZWViYmZhMDYyIyM3Mzc2MWYwNi0yYWM5LTQ2OWMtOWYxMC0yNzlhOGNjMjY3Zjk="
@pytest.mark.parametrize(
"channel_type, expected_exception, mock_get_membership_id, expected_error_value",
[
("private", None, membership_id, None),
("shared", None, membership_id, None),
("standard", ValueError, None, "Removing a member is allowed only for private or shared channels."),
("shared", ValueError, "", 'User "itayadmin" was not found in channel "test channel".'),
],
)
def test_user_remove_from_channel_command(
mocker, requests_mock, channel_type, expected_exception, mock_get_membership_id, expected_error_value
):
"""
Given:
- The commands arguments
When:
- Executing the 'microsoft-teams-user-remove-from-channel' command.
Then:
- Verify human-readable output
- Verify the expected error is raised.
"""
from MicrosoftTeams import user_remove_from_channel_command
mocker.patch.object(
demisto, "args", return_value={"channel_name": "test channel", "team": "test team", "member": "itayadmin"}
)
mocker.patch("MicrosoftTeams.get_team_aad_id", return_value=team_aad_id)
mocker.patch("MicrosoftTeams.get_channel_id", return_value=channel_id)
mocker.patch("MicrosoftTeams.get_channel_type", return_value=channel_type)
mocker.patch("MicrosoftTeams.get_user_membership_id", return_value=mock_get_membership_id)
return_results = mocker.patch("MicrosoftTeams.return_results")
requests_mock.delete(
f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels/{channel_id}/members/{membership_id}", status_code=204
)
if expected_exception:
with pytest.raises(ValueError) as e:
user_remove_from_channel_command()
assert str(e.value) == expected_error_value
else:
user_remove_from_channel_command()
assert requests_mock.request_history[0].method == "DELETE"
results_hr = return_results.call_args[0][0]
assert results_hr == 'The user "itayadmin" has been removed from channel "test channel" successfully.'
# All the responses based on: https://learn.microsoft.com/en-us/graph/api/
test_data = load_test_data("./test_data/chats_test_data.json")
def test_chat_create_command(mocker):
"""
Given:
- The command arguments to create group chat
When:
- Executing the 'microsoft-teams-chat-create' command.
Then:
- Verify human-readable output
- Verify entry context output
"""
from MicrosoftTeams import chat_create_command
mocker.patch.object(
demisto, "args", return_value={"chat_type": "group", "chat_name": "Group_chat", "member": "testuser1@example.com"}
)
api_response = test_data.get("create_group_chat")
expected_hr_create_chat = test_data.get("expected_hr_create_chat")
return_results = mocker.patch("MicrosoftTeams.return_results")
mocker.patch("MicrosoftTeams.get_user", return_value=[{"id": "user1", "userType": "Member"}])
mocker.patch("MicrosoftTeams.create_chat", return_value=api_response)
mocker.patch("MicrosoftTeams.add_bot_to_chat", return_value="")
chat_create_command()
results_hr = return_results.call_args[0][0].readable_output
results_outputs = return_results.call_args[0][0].outputs
assert results_hr == expected_hr_create_chat
api_response.pop("@odata.context", "")
assert results_outputs == api_response
@pytest.mark.parametrize(
"chat_type, users, expected_request_json, chat_response",
[
(
"group",
[("8b081ef6-4792-4def-b2c9-c363a1bf41d5", "Member"), ("82af01c5-f7cc-4a2e-a728-3a5df21afd9d", "Guest")],
"group_request_json",
"create_group_chat",
),
("oneOnOne", [("8b081ef6-4792-4def-b2c9-c363a1bf41d5", "Member")], "oneOnOne_request_json", "create_oneOnOne_chat"),
],
)
def test_creat_chat(mocker, requests_mock, chat_type, users, expected_request_json, chat_response):
"""
Given:
- The function arguments chat_type, users
When:
- Calling the create_chat function
Then:
- Ensure expected request body is sent
"""
from MicrosoftTeams import create_chat
signed_in_response = test_data.get("signed_in_user")
mocker.patch("MicrosoftTeams.get_signed_in_user", return_value=signed_in_response)
requests_mock.post("https://graph.microsoft.com/v1.0/chats", json=chat_response)
create_chat(chat_type, users, "Group chat title") # the chat name will not be used in oneOnOne chats - only in group
assert requests_mock.request_history[0].json() == test_data.get(expected_request_json)
def test_message_send_to_chat_command(mocker, requests_mock):
"""
Given:
- The command arguments
When:
- Executing the 'microsoft-teams-message-send-to-chat' command.
Then:
- Assert the request url is as expected
- Verify human-readable output
- Verify entry context output
"""
from MicrosoftTeams import message_send_to_chat_command
mocker.patch.object(demisto, "args", return_value={"content": "Hello World"})
return_results = mocker.patch("MicrosoftTeams.return_results")
mock_response = test_data.get("send_message_chat")
mocker.patch("MicrosoftTeams.get_chat_id_and_type", return_value=(GROUP_CHAT_ID, "group"))
mocker.patch("MicrosoftTeams.add_bot_to_chat", return_value="")
requests_mock.post(f"{GRAPH_BASE_URL}/v1.0/chats/{GROUP_CHAT_ID}/messages", json=mock_response)
message_send_to_chat_command()
assert requests_mock.request_history[0].json() == {
"body": {"content": "Hello World", "contentType": "text"},
"messageType": "message",
}
results_hr = return_results.call_args[0][0].readable_output
results_outputs = return_results.call_args[0][0].outputs
assert results_hr == test_data.get("expected_hr_send_message")
mock_response.pop("@odata.context", "")
expected_outputs = {"chatId": GROUP_CHAT_ID, "messages": mock_response}
assert results_outputs == expected_outputs
def test_chat_member_list_command(mocker, requests_mock):
"""
Given:
- The command arguments
When:
- Executing the 'microsoft-teams-chat-member-list' command.
Then:
- Assert the request url is as expected
- Verify human-readable output
- Verify entry context output
"""
from MicrosoftTeams import chat_member_list_command
mocker.patch.object(demisto, "args", return_value={"chat": ONEONONE_CHAT_ID})
return_results = mocker.patch("MicrosoftTeams.return_results")
mock_response = test_data.get("list_members")
expected_hr_chat_member_list = test_data.get("expected_hr_chat_member_list")
get_chat_members_expected_response = mock_response.get("value", [])
mocker.patch("MicrosoftTeams.get_chat_id_and_type", return_value=(ONEONONE_CHAT_ID, "oneOnOne"))
requests_mock.get(f"{GRAPH_BASE_URL}/v1.0/chats/{ONEONONE_CHAT_ID}/members", json=mock_response)
chat_member_list_command()
results_hr = return_results.call_args[0][0].readable_output
results_outputs = return_results.call_args[0][0].outputs
assert results_hr == expected_hr_chat_member_list
[member.pop("@odata.type", None) for member in get_chat_members_expected_response]
expected_outputs = {"chatId": ONEONONE_CHAT_ID, "members": get_chat_members_expected_response}
assert results_outputs == expected_outputs
@pytest.mark.parametrize(
"chat_id, chat_type, expected_exception", [(GROUP_CHAT_ID, "group", False), (ONEONONE_CHAT_ID, "oneOnOne", True)]
)
def test_chat_update_command(mocker, requests_mock, chat_id, chat_type, expected_exception):
"""
Given:
- The command arguments:
- GROUP_CHAT_ID, group -> updates the title
- ONEONONE_CHAT_ID, oneOnOne -> raise ValueError
When:
- Executing the 'microsoft-teams-chat-update' command.
Then:
- Assert the request url is as expected
- Verify human-readable output
"""
from MicrosoftTeams import chat_update_command
mocker.patch.object(demisto, "args", return_value={"chat": chat_id, "chat_name": "XsoarChat"})
mocker.patch("MicrosoftTeams.get_chat_id_and_type", return_value=(chat_id, chat_type))
return_results = mocker.patch("MicrosoftTeams.return_results")
mock_response = test_data.get("update_chat")
requests_mock.patch(f"{GRAPH_BASE_URL}/v1.0/chats/{chat_id}", json=mock_response)
if expected_exception:
with pytest.raises(ValueError) as e:
chat_update_command()
assert str(e.value) == "Setting chat name is allowed only on group chats."
else:
chat_update_command()
assert requests_mock.request_history[0].method == "PATCH"
assert requests_mock.request_history[0].json() == {"topic": "XsoarChat"}
results_hr = return_results.call_args[0][0]
assert results_hr == f"The name of chat '{chat_id}' has been successfully changed to 'XsoarChat'."
@pytest.mark.parametrize(
"chat, member, expected_exit, expected_warning, expected_result, mocked_get_user",
[
(
"group",
"user1",
False,
None,
'The User "user1" has been added to chat "group" successfully.',
[[{"id": 1, "userType": "Member"}]],
),
(
"group",
"user1,user2",
False,
None,
'The Users "user1, user2" have been added to chat "group" successfully.',
[[{"id": 1, "userType": "Member"}], [{"id": 2, "userType": "Member"}]],
),
(
"group",
"user1,unknown",
False,
"The following members were not found: unknown",
'The User "user1" has been added to chat "group" successfully.',
[[{"id": 1, "userType": "Member"}], []],
),
("group", "unknown1,unknown2", True, "The following members were not found: unknown1, unknown2", None, [[], []]),
("oneOnOne", "user1", True, ValueError, "Adding a member is allowed only on group chat.", []),
],
)
def test_chat_add_user_command(mocker, chat, member, expected_exit, expected_warning, expected_result, mocked_get_user):
"""
Given:
- Adding a single member to a group chat
- Adding multiple members to a group chat
- Adding a non-existing member to a group chat
- Adding multiple non-existing members to a group chat
- Adding a member to a oneOnOne chat
When:
- Executing the 'microsoft-teams-chat-add-user' command.
Then:
- verify that the relevant functions are called correctly based on the inputs and expected results.
- checks if the correct warning or result is returned, or if the correct exception is raised,
based on the inputs and expected outcomes.
"""
import MicrosoftTeams
from MicrosoftTeams import chat_add_user_command
mocker.patch.object(demisto, "args", return_value={"chat": chat, "member": member})
mocker.patch.object(demisto, "results")
warning = mocker.patch.object(MicrosoftTeams, "return_warning")
get_chat_id_and_type_mock = mocker.patch("MicrosoftTeams.get_chat_id_and_type", return_value=(chat, chat))
get_user_mock = mocker.patch("MicrosoftTeams.get_user", side_effect=mocked_get_user)
add_user_to_chat_mock = mocker.patch("MicrosoftTeams.add_user_to_chat")
if expected_warning is ValueError:
with pytest.raises(ValueError) as e:
chat_add_user_command()
assert str(e.value) == expected_result
else:
chat_add_user_command()
if expected_warning:
warning.assert_called_once_with(expected_warning, exit=expected_exit)
if expected_result:
demisto.results.assert_called_once_with(expected_result)
get_chat_id_and_type_mock.assert_called_once_with(chat)
if not expected_warning:
get_user_mock.assert_called()
add_user_to_chat_mock.assert_called()
def test_add_user_to_chat(requests_mock):
"""
Given:
- The function arguments
When:
- Calling the add_user_to_chats function
Then:
- Ensure expected request body is sent
"""
from MicrosoftTeams import add_user_to_chat
expected_request_json = test_data.get("add_member_request")
requests_mock.post(f"https://graph.microsoft.com/v1.0/chats/{GROUP_CHAT_ID}/members", status_code=201)
add_user_to_chat(GROUP_CHAT_ID, "Member", "8b081ef6-4792-4def-b2c9-c363a1bf41d5", True)
assert requests_mock.request_history[0].json() == expected_request_json
@pytest.mark.parametrize(
"args, expected_response, expected_request_url, expected_outputs",
[
({"chat": "test group 1", "filter": "test_filter"}, ValueError, "", ""),
(
{"chat": "test group 1"},
"get_chat",
f"https://graph.microsoft.com/v1.0/chats/{GROUP_CHAT_ID}",
"expected_outputs_get_chat",
),
(
{"expand": "members", "limit": 3},
"list_chats_with_members",
"https://graph.microsoft.com/v1.0/chats/?%24expand=members&%24top=3",
"expected_outputs_list_chats_with_members",
),
(
{"expand": "lastMessagePreview", "limit": 3},
"list_chats_with_lastMessagePreview",
"https://graph.microsoft.com/v1.0/chats/?%24expand=lastMessagePreview&%24top=3",
"expected_outputs_list_chats_with_lastMessagePreview",
),
({"limit": 3}, "list_chats", "https://graph.microsoft.com/v1.0/chats/?%24top=3", "expected_outputs_list_chats"),
({}, "list_chats", "https://graph.microsoft.com/v1.0/chats/?%24top=50", "expected_outputs_list_chats"),
(
{"next_link": "https://graph.microsoft.com/v1.0/chats/test_next_link", "page_size": 3},
"list_chats",
"https://graph.microsoft.com/v1.0/chats/test_next_link",
"expected_outputs_list_chats",
),
],
)
def test_chat_list_command(mocker, requests_mock, args, expected_response, expected_request_url, expected_outputs):
"""
Given:
- The command arguments
When:
- Executing the 'microsoft-teams-chat-list' command.
Then:
- Assert the request url is as expected
- Verify that the context outputs is as expected
"""
from MicrosoftTeams import chat_list_command
mocker.patch("MicrosoftTeams.get_chat_id_and_type", return_value=(GROUP_CHAT_ID, "group"))
return_results = mocker.patch("MicrosoftTeams.return_results")
mocker.patch.object(demisto, "args", return_value=args)
if expected_response is ValueError:
with pytest.raises(ValueError) as e:
chat_list_command()
assert str(e.value) == "Retrieve a single chat does not support the 'filter' ODate query parameter."
else:
requests_mock.get(expected_request_url, json=test_data.get(expected_response))
chat_list_command()
assert return_results.call_args[0][0].outputs == test_data.get(expected_outputs)
@pytest.mark.parametrize(
"args, expected_response, expected_request_url, expected_outputs",
[
(
{"limit": 2, "order_by": "createdDateTime"},
"list_messages",
f"https://graph.microsoft.com/v1.0/chats/{GROUP_CHAT_ID}/messages?$top=2&$orderBy=createdDateTime desc",
"expected_outputs_list_messages",
),
(
{"next_link": "https://graph.microsoft.com/v1.0/chats/test_next_link", "page_size": 2},
"list_messages",
"https://graph.microsoft.com/v1.0/chats/test_next_link",
"expected_outputs_list_messages",
),
],
)
def test_chat_message_list_command(mocker, requests_mock, args, expected_response, expected_request_url, expected_outputs):
"""
Given:
- The command arguments
When:
- Executing the 'microsoft-teams-chat-message-list' command.
Then:
- Assert the request url is as expected
- Verify that the context outputs is as expected
"""
from MicrosoftTeams import chat_message_list_command
mocker.patch("MicrosoftTeams.get_chat_id_and_type", return_value=(GROUP_CHAT_ID, "group"))
return_results = mocker.patch("MicrosoftTeams.return_results")
mocker.patch.object(demisto, "args", return_value=args)
requests_mock.get(expected_request_url, json=test_data.get(expected_response))
chat_message_list_command()
assert return_results.call_args[0][0].outputs == test_data.get(expected_outputs)
def test_pages_puller(requests_mock):
"""
Given:
- The function arguments: response, limit:
- The response has a nextLink URL.
- The limit is greater than the number of results in the first response.
When:
- Calling the 'pages_puller' function.
Then:
- Assert the request url is as expected - make an API request using the nextLink URL.
- Verify the function output is as expected
"""
from MicrosoftTeams import pages_puller
response = test_data.get("list_messages") # contains 2 results
limit = 4
expected_result = response.get("value") * 2
requests_mock.get(response.get("@odata.nextLink"), json=response)
result, last_next_link = pages_puller(response, limit)
assert requests_mock.call_count == 1
assert result == expected_result
assert last_next_link == response.get("@odata.nextLink")
def test_get_chat_id_and_type(mocker, requests_mock):
"""
Given:
The 'chat' argument as:
- case 1: chat ID -> returns the given ID and the chat_type
- case 2: chat_name (topic) -> returns the ID and 'group' chat_type
- case 3: member -> returns the ID of a one-on-one chat and 'oneOnOne' chat_type
- case 4: non-existing member/chat_name (topic) -> raise ValueError
When:
- Calling the 'get_chat_id_and_type' function.
Then:
- Assert the request url is as expected
- Verify the function output is as expected
"""
from MicrosoftTeams import get_chat_id_and_type
# case 1: chat = chat_id [= GROUP_CHAT_ID]
requests_mock.get(f"{GRAPH_BASE_URL}/v1.0/chats/{GROUP_CHAT_ID}", json=test_data.get("get_chat"))
assert get_chat_id_and_type(GROUP_CHAT_ID) == (GROUP_CHAT_ID, "group")
# case 2: chat = chat_name (topic) [= "test"]
requests_mock.get(
"https://graph.microsoft.com/v1.0/chats/?$select=id, chatType&$filter=topic eq 'test'",
json=test_data.get("get_chat_id_and_type_response"),
)
assert get_chat_id_and_type("test") == (GROUP_CHAT_ID, "group")
# case 3: chat = member [= "test_admin"]
requests_mock.get(
"https://graph.microsoft.com/v1.0/chats/?$select=id, chatType&$filter=topic eq 'test_admin'",
json=test_data.get("get_chat_id_and_type_no_chat_response"),
)
get_user_mock = mocker.patch("MicrosoftTeams.get_user", return_value=[{"id": "user_id", "userType": "Member"}])
create_chat_mock = mocker.patch("MicrosoftTeams.create_chat", return_value=test_data.get("create_oneOnOne_chat"))
assert get_chat_id_and_type("test_admin") == (
"19:82fe7758-5bb3-4f0d-a43f-e555fd399c6f_8c0a1a67-50ce-4114-bb6c-da9c5dbcf6ca@unq.gbl.spaces",
"oneOnOne",
)
assert create_chat_mock.call_args.args == ("oneOnOne", [("user_id", "Member")])
assert get_user_mock.call_count == 1
assert create_chat_mock.call_count == 1
# case 4: chat = non-existing member or chat_name (topic) [= "unknown"]
requests_mock.get(
"https://graph.microsoft.com/v1.0/chats/?$select=id, chatType&$filter=topic eq 'unknown'",
json=test_data.get("get_chat_id_and_type_no_chat_response"),
)
get_user_mock = mocker.patch("MicrosoftTeams.get_user", return_value=[])
with pytest.raises(ValueError) as e:
get_chat_id_and_type("unknown")
assert str(e.value) == "Could not find chat: unknown"
def test_generate_login_url(mocker):
"""
Given:
- Self-deployed are true and auth code are the auth flow
When:
- Calling function microsoft-teams-generate-login-url
Then:
- Ensure the generated url are as expected.
"""
# prepare
import demistomock as demisto
import MicrosoftTeams
from MicrosoftTeams import main
redirect_uri = "redirect_uri"
tenant_id = "tenant_id"
client_id = "client_id"
mocked_params = {"REDIRECT_URI": redirect_uri, "AUTH_TYPE": "Authorization Code", "BOT_ID": client_id}
mocker.patch.dict(MicrosoftTeams.__dict__, MicrosoftTeams.__dict__ | mocked_params)
mocker.patch.object(MicrosoftTeams, "get_integration_context", return_value={"tenant_id": "tenant_id"})
mocker.patch.object(MicrosoftTeams, "return_results")
mocker.patch.object(MicrosoftTeams, "support_multithreading")
mocker.patch.object(demisto, "command", return_value="microsoft-teams-generate-login-url")
# call
main()
# assert
expected_url = (
f"[login URL](https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/authorize?"
"response_type=code&scope=offline_access%20https://graph.microsoft.com/.default"
f"&client_id={client_id}&redirect_uri={redirect_uri})"
)
res = MicrosoftTeams.return_results.call_args[0][0].readable_output
assert expected_url in res
def test_is_bot_in_chat_parameters(mocker, requests_mock):
"""
Given: some chat ID and bot ID
When: calling is_bot_in_chat() to check if the bot is already a member of the chat
Then: validate that the request is sent correctly and specifically that the BOT_ID is part of the query
"""
request_mock = requests_mock.get(f"{GRAPH_BASE_URL}/v1.0/chats/{GROUP_CHAT_ID}/installedApps", json={})
mocker.patch("MicrosoftTeams.BOT_ID", new=bot_id)
from MicrosoftTeams import is_bot_in_chat
is_bot_in_chat(GROUP_CHAT_ID)
filters = request_mock.last_request.qs.get("$filter")[0]
assert f"eq '{bot_id}'" in filters
@pytest.mark.parametrize(
"error_content, status_code, expected_response",
[
(
b'{"error": "invalid_grant", "error_description": "AADSTS700082: The refresh token has expired due to inactivity.'
b"\\u00a0The token was issued on 2023-02-06T12:26:14.6448497Z and was inactive for 90.00:00:00."
b'\\r\\nTrace ID: test\\r\\nCorrelation ID: test\\r\\nTimestamp: 2023-07-02 06:40:26Z", '
b'"error_codes": [700082], "timestamp": "2023-07-02 06:40:26Z", "trace_id": "test", "correlation_id": "test",'
b' "error_uri": "https://login.microsoftonline.com/error?code=700082"}',
400,
"The refresh token has expired due to inactivity. Please regenerate the "
"'Authorization code' parameter and then run !microsoft-teams-auth-test to "
"re-authenticate",
)
],
)
def test_error_parser_with_exception(mocker, error_content, status_code, expected_response):
"""
Given:
- The error_content, status_code, and expected_response for testing the error_parser function.
When:
- The error_parser function is called with the given error_content and status_code.
Then:
- Assert that the error_parser function raises a DemistoException with the expected_response.
"""
from MicrosoftTeams import error_parser
mocker.patch.object(demisto, "getIntegrationContext", return_value=integration_context)
mocker.patch.object(demisto, "setIntegrationContext")
mocker.patch.object(demisto, "error")
err = Response()
err.status_code = status_code
err._content = error_content
with pytest.raises(DemistoException) as ex:
error_parser(err)
assert demisto.getIntegrationContext.call_count == 1
assert demisto.setIntegrationContext.call_count == 1
assert str(ex.value) == expected_response
@pytest.mark.parametrize(
"is_xsoar_8, expected_result", [(True, ({"http": "xsoar_8_proxy", "https": "xsoar_8_proxy"}, True)), (False, (None, False))]
)
def test_handle_teams_proxy_and_ssl(mocker, is_xsoar_8, expected_result):
"""
Given:
- If the xsoar version is greater or less than 8, and the expected reuslts of the integration proxies.
When:
- The version of xsoar is greater than 8 or less than 8
Then:
- Assert that when the version is greater than 8, proxies dict is not empty and use_ssl is true
- Assert that when the version is less than 8, proxies dict is empty and use_ssl is false.
"""
import MicrosoftTeams as ms_teams
os.environ["CRTX_HTTP_PROXY"] = "xsoar_8_proxy"
mocker.patch.object(ms_teams, "is_demisto_version_ge", return_value=is_xsoar_8)
mocker.patch.object(ms_teams, "PARAMS", return_value={"insecure": True})
proxies, use_ssl = ms_teams.handle_teams_proxy_and_ssl()
assert (proxies, use_ssl) == expected_result
DUMMY_ASK_MESSAGE = {
"message_text": "message",
"options": ["option"],
"entitlement": "id",
"investigation_id": "inv_id",
"task_id": "task",
"form_type": "form",
}
@pytest.mark.parametrize(
"message, result",
[
(json.dumps(DUMMY_ASK_MESSAGE), True),
(json.dumps(DUMMY_ASK_MESSAGE | {"extra_key": "extra"}), False),
("non json message", False),
],
)
def test_is_teams_ask_message(message, result):
"""
Given:
- input message string
When:
- Running is_teams_ask_message.
Then:
- Assert only ask_teams messages return True
If the test fails, please update the first message param in the test to have the same keys as MS_TEAMS_ASK_MESSAGE_KEYS
constant in MicrosoftTeams.
"""
from MicrosoftTeams import is_teams_ask_message
assert is_teams_ask_message(message) == result
def test_add_data_to_actions_simple_card():
from MicrosoftTeams import add_data_to_actions
card_json = {"type": "Action.Submit", "title": "Submit"}
data_value = {"key": "value"}
add_data_to_actions(card_json, data_value)
assert card_json["data"] == data_value
def test_add_data_to_actions_nested_card():
from MicrosoftTeams import add_data_to_actions
card_json = {
"type": "AdaptiveCard",
"actions": [{"type": "Action.Submit", "title": "Submit 1"}, {"type": "Action.Execute", "title": "Execute 1"}],
}
data_value = {"key": "value"}
add_data_to_actions(card_json, data_value)
assert card_json["actions"][0]["data"] == data_value
assert card_json["actions"][1]["data"] == data_value
def test_add_data_to_actions_show_card():
from MicrosoftTeams import add_data_to_actions
card_json = {
"type": "Action.ShowCard",
"title": "Show Card",
"card": {"type": "AdaptiveCard", "actions": [{"type": "Action.Submit", "title": "Nested Submit"}]},
}
data_value = {"key": "value"}
add_data_to_actions(card_json, data_value)
assert card_json["card"]["actions"][0]["data"] == data_value
def test_add_data_to_actions_mixed_types():
from MicrosoftTeams import add_data_to_actions
card_json = [
{"type": "Action.Submit", "title": "Submit"},
{"type": "TextBlock", "text": "Some text"},
{"type": "Action.Execute", "title": "Execute"},
]
data_value = {"key": "value"}
add_data_to_actions(card_json, data_value)
assert card_json[0]["data"] == data_value
assert "data" not in card_json[1]
assert card_json[2]["data"] == data_value
def test_add_data_to_actions_empty_input():
from MicrosoftTeams import add_data_to_actions
card_json = {}
data_value = {"key": "value"}
add_data_to_actions(card_json, data_value)
assert card_json == {}
def test_add_data_to_actions_non_dict_data():
from MicrosoftTeams import add_data_to_actions
card_json = {"type": "Action.Submit", "title": "Submit"}
data_value = "string_data"
add_data_to_actions(card_json, data_value)
assert card_json["data"] == data_value
@pytest.mark.parametrize(
"token, decoded_token, auth_type, expected_hr",
[
(
"dummy_token",
{"aud": "url", "exp": "1111", "roles": ["AppCatalog.Read.All", "Group.ReadWrite.All", "User.Read.All"]},
"Client Credentials",
"The current API permissions in the Teams application are",
),
(
"dummy_token",
{"aud": "url", "exp": "1111", "roles": []},
"Client Credentials",
"No permissions obtained for the used graph access token.",
),
(
"dummy_token",
{"aud": "url", "exp": "1111", "scp": "AppCatalog.Read.All Group.ReadWrite.All User.Read.All"},
"Authorization Code",
"The current API permissions in the Teams application are",
),
(
"dummy_token",
{"aud": "url", "exp": "1111", "scp": ""},
"Authorization Code",
"No permissions obtained for the used graph access token.",
),
("", {"roles": []}, "Client Credentials", "Graph access token is not set."),
],
ids=[
"Test token permissions list command - client credentials auth flow",
"Test token permissions list command - client credentials auth flow - no permissions set",
"Test token permissions list command - auth code auth flow",
"Test token permissions list command - client auth code flow - no permissions set",
"Test token permissions list command - missing token",
],
)
def test_token_permissions_list_command(mocker, token, decoded_token, auth_type, expected_hr):
"""
Tests the 'token_permissions_list_command' logic:
For client credentials auth flow, the API permissions are found under the 'roles' key in the decoded token data,
while for the auth code flow they are found under the 'scp' key.
This test checks that we extract the API permissions from the graph access token successfully for both auth types.
Given:
1. A dummy token, mocked response of the jet.decode func with API permissions roles under the 'roles' key -
(auth type is client credentials).
2. A dummy token, mocked response of the jet.decode func without API permissions roles under the 'roles' key -
(auth type is client credentials).
3. A dummy token, mocked response of the jet.decode func with API permissions roles under the 'scp' key -
(auth type is Authorization Code).
4. A dummy token, mocked response of the jet.decode func without API permissions roles under the 'scp' key -
(auth type is Authorization Code).
5. Missing token.
When:
- Running the token_permissions_list_command.
Then:
Verify that the human readable output is as expected:
1. API permissions list.
2. No permissions obtained for the used graph access token.
3. API permissions list.
4. No permissions obtained for the used graph access token.
5. Graph access token is not set.
"""
import MicrosoftTeams
from MicrosoftTeams import token_permissions_list_command
mocker.patch("MicrosoftTeams.get_graph_access_token", return_value=token)
mocker.patch("MicrosoftTeams.AUTH_TYPE", new=auth_type)
mocker.patch("jwt.decode", return_value=decoded_token)
results = mocker.patch.object(MicrosoftTeams, "return_results")
token_permissions_list_command()
assert expected_hr in results.call_args[0][0].readable_output
@pytest.mark.parametrize(
"xsoar_server, is_xsoar_on_prem, is_xsiam, expected_hr",
[
("https://dns-test.name:443", True, False, "https://dns-test.name:443/instance/execute/teams"),
(
"https://viso-test-dummy.crtx-qa-ttt.ss.paloaltonetworks.com",
False,
False,
"https://ext-viso-test-dummy.crtx-qa-ttt.ss.paloaltonetworks.com/xsoar/instance/execute/teams",
),
(
"http://viso-test-dummy.crtx-qa-ttt.ss.paloaltonetworks.com",
False,
False,
"http://ext-viso-test-dummy.crtx-qa-ttt.ss.paloaltonetworks.com/xsoar/instance/execute/teams",
),
(
"https://viso-test-dummy.xdr-qa-ttt.ss.paloaltonetworks.com",
False,
True,
"https://ext-viso-test-dummy.crtx-qa-ttt.ss.paloaltonetworks.com/xsoar/instance/execute/teams",
),
(
"http://viso-test-dummy.xdr.qa-ttt.ss.paloaltonetworks.com",
False,
True,
"http://ext-viso-test-dummy.crtx.qa-ttt.ss.paloaltonetworks.com/xsoar/instance/execute/teams",
),
(
"http://viso-test-dummy.crtx-qa-ttt.ss.paloaltonetworks.com",
False,
True,
"http://ext-viso-test-dummy.crtx-qa-ttt.ss.paloaltonetworks.com/xsoar/instance/execute/teams",
),
],
ids=[
"Test xsoar 6 server url",
"Test xsoar 8 server url (with https:// prefix)",
"Test xsoar 8 server url (with http:// prefix)",
"Test xsiam server url (with https:// prefix)",
"Test xsiam server url (with http:// prefix)",
"Test xsiam server url without the 'xdr' string in the dns name",
],
)
def test_create_messaging_endpoint_command(mocker, xsoar_server, is_xsoar_on_prem, is_xsiam, expected_hr):
"""
Tests the 'create_messaging_endpoint_command' logic.
Given:
1. An xsoar 6 server url.
2. An xsoar 8 server url (with https:// prefix).
3. An xsoar 8 server url (with http:// prefix).
4. An xsiam server url (with https:// prefix).
5. An xsiam server url (with http:// prefix).
6. An xsiam server url without the 'xdr' string in the dns name.
When:
- Running the create_messaging_endpoint_command.
Then:
Verify that the messaging endpoint was created as expected:
1. The 'instance/execute/teams' suffix was added.
2. The 'ext' prefix was added to the dns name, and the 'xsoar/instance/execute/teams' suffix was added.
3. The 'ext' prefix was added to the dns name, and the 'xsoar/instance/execute/teams' suffix was added.
4. The 'ext' prefix was added to the dns name, the 'xdr' was replaced with 'crtx' and the 'xsoar/instance/execute/teams'
suffix was added.
5. The 'ext' prefix was added to the dns name, the 'xdr' was replaced with 'crtx' and the 'xsoar/instance/execute/teams'
suffix was added.
6. The 'ext' prefix was added to the dns name, and the 'xsoar/instance/execute/teams' suffix was added.
"""
import MicrosoftTeams
from MicrosoftTeams import create_messaging_endpoint_command
mocker.patch.object(demisto, "demistoUrls", return_value={"server": xsoar_server})
mocker.patch.object(demisto, "integrationInstance", return_value="teams")
mocker.patch.object(demisto, "args", return_value={"engine_url": ""})
mocker.patch("MicrosoftTeams.is_xsoar_on_prem", return_value=is_xsoar_on_prem)
mocker.patch("MicrosoftTeams.is_xsiam", return_value=is_xsiam)
mocker.patch("MicrosoftTeams.is_using_engine", return_value=False)
results = mocker.patch.object(MicrosoftTeams, "return_results")
create_messaging_endpoint_command()
assert expected_hr in results.call_args[0][0].readable_output
@pytest.mark.parametrize(
"engine_url, is_xsoar_on_prem, is_xsiam, expected_hr",
[
("https://my-engine.com:333", True, False, "https://my-engine.com:333"),
("https://my-engine.com:333", False, False, "https://my-engine.com:333"),
("https://my-engine.com:333", False, True, "https://my-engine.com:333"),
("https://1.1.1.1:333", False, True, "https://1.1.1.1:333"),
],
ids=[
"Test xsoar 6 engine url",
"Test xsoar 8 engine url",
"Test xsiam engine url",
"Test xsoar engine url - with IP",
],
)
def test_create_messaging_endpoint_command_for_xsoar_engine(mocker, engine_url, is_xsoar_on_prem, is_xsiam, expected_hr):
"""
Tests the 'create_messaging_endpoint_command' logic when the user uses an xsoar engine.
Given:
- An xsoar engine url.
When:
- Running the create_messaging_endpoint_command on:
1. xsoar 6
2. xsoar 8
3. xsiam
4. The engine url include an IP and not a DNS name.
Then:
Verify that the messaging endpoint was created as expected - only the engine url and port (without any suffix).
"""
import MicrosoftTeams
from MicrosoftTeams import create_messaging_endpoint_command
mocker.patch.object(demisto, "demistoUrls", return_value={"server": "https://test-server.com:443"})
mocker.patch.object(demisto, "integrationInstance", return_value="teams")
mocker.patch.object(demisto, "args", return_value={"engine_url": engine_url})
mocker.patch("MicrosoftTeams.is_xsoar_on_prem", return_value=is_xsoar_on_prem)
mocker.patch("MicrosoftTeams.is_xsiam", return_value=is_xsiam)
mocker.patch("MicrosoftTeams.is_using_engine", return_value=True)
results = mocker.patch.object(MicrosoftTeams, "return_results")
create_messaging_endpoint_command()
assert expected_hr in results.call_args[0][0].readable_output
@pytest.mark.parametrize(
"engine_url",
[
("https://my-engine.com"),
("my-engine.com:333"),
("https://my engine.com:433"),
],
ids=[
"Test engine url without a port",
"Test engine url without an http or https prefix",
"Test engine url with spaces in the dns name",
],
)
def test_create_messaging_endpoint_command_invalid_xsoar_engine(mocker, engine_url):
"""
Tests the 'create_messaging_endpoint_command' logic when the user uses an xsoar engine, and provides an invalid engine url.
Given:
- An invalid engine URL:
1. without a port.
2. without an http:// or https:// prefix
3. with a space in the dns name
When:
- Running the create_messaging_endpoint_command.
Then:
Verify that a valueError exception is raised with the error description.
"""
import MicrosoftTeams
from MicrosoftTeams import create_messaging_endpoint_command
mocker.patch.object(demisto, "demistoUrls", return_value={"server": "https://test-server.com:443"})
mocker.patch.object(demisto, "integrationInstance", return_value="teams")
mocker.patch.object(demisto, "args", return_value={"engine_url": engine_url})
mocker.patch("MicrosoftTeams.is_using_engine", return_value=True)
mocker.patch.object(MicrosoftTeams, "return_results")
with pytest.raises(ValueError) as e:
create_messaging_endpoint_command()
assert "Invalid engine URL -" in str(e.value)
def test_switch_auth_type_to_client_credentials(mocker):
"""
Tests the 'auth_type_switch_handling' logic when the user switched the auth type in the instance parameters from Auth Code
Flow to the Client Credentials Flow.
Given:
- Auth type instance parameter is now 'Client Credentials'.
When:
- Running the 'auth_type_switch_handling' function.
Then:
- Verify that the integration context was updated as follows:
1. current_auth_type = 'Client Credentials'.
2. graph token related values were deleted.
- Verify that the debug logs are as expected.
"""
from MicrosoftTeams import auth_type_switch_handling
mocker.patch(
"MicrosoftTeams.get_integration_context",
return_value={
"current_auth_type": "Authorization Code",
AUTHCODE_TOKEN_PARAMS: json.dumps(
{
"current_refresh_token": "test_refresh_token",
"graph_access_token": "test_graph_token",
"graph_valid_until": "test_valid_until",
}
),
},
)
set_integration_context_mocker = mocker.patch("MicrosoftTeams.set_integration_context", return_value={})
debug_log_mocker = mocker.patch.object(demisto, "debug")
mocker.patch("MicrosoftTeams.AUTH_TYPE", new="Client Credentials")
auth_type_switch_handling()
assert set_integration_context_mocker.call_count == 2
assert set_integration_context_mocker.call_args[0][0] == {
"current_auth_type": "Client Credentials",
AUTHCODE_TOKEN_PARAMS: "{}",
CREDENTIALS_TOKEN_PARAMS: "{}",
}
assert "Setting the current_auth_type in the integration context to Client Credentials" in debug_log_mocker.call_args[0][0]
assert debug_log_mocker.call_count == 4
def test_switch_auth_type_to_authorization_code_flow(mocker):
"""
Tests the 'auth_type_switch_handling' logic when the user switched the auth type in the instance parameters from the
Client Credentials Flow to the Auth Code Flow.
Given:
- Auth type instance parameter is now 'Authorization Code'.
When:
- Running the 'auth_type_switch_handling' function.
Then:
- Verify that the integration context was updated as follows:
1. current_auth_type = 'Authorization Code'.
2. graph token related values were deleted.
- Verify that the debug logs are as expected.
"""
from MicrosoftTeams import auth_type_switch_handling
mocker.patch(
"MicrosoftTeams.get_integration_context",
return_value={
"current_auth_type": "Client Credentials",
AUTHCODE_TOKEN_PARAMS: json.dumps(
{
"current_refresh_token": "test_refresh_token",
"graph_access_token": "test_graph_token",
"graph_valid_until": "test_valid_until",
}
),
},
)
set_integration_context_mocker = mocker.patch("MicrosoftTeams.set_integration_context", return_value={})
debug_log_mocker = mocker.patch.object(demisto, "debug")
mocker.patch("MicrosoftTeams.AUTH_TYPE", new="Authorization Code")
auth_type_switch_handling()
assert set_integration_context_mocker.call_count == 2
assert set_integration_context_mocker.call_args[0][0] == {
"current_auth_type": "Authorization Code",
AUTHCODE_TOKEN_PARAMS: "{}",
CREDENTIALS_TOKEN_PARAMS: "{}",
}
assert "Setting the current_auth_type in the integration context to Authorization Code" in debug_log_mocker.call_args[0][0]
assert debug_log_mocker.call_count == 4
def test_auth_type_handling_for_first_run_of_the_instance(mocker):
"""
Tests the 'auth_type_switch_handling' logic in the first run of the integration instance/
Given:
- Auth type instance parameter is now 'Authorization Code'.
When:
- Running the 'auth_type_switch_handling' function.
Then:
- Verify that the integration context was updated as follows:
1. current_auth_type = 'Authorization Code'.
- Verify that the debug logs are as expected.
"""
from MicrosoftTeams import auth_type_switch_handling
mocker.patch("MicrosoftTeams.get_integration_context", return_value={})
set_integration_context_mocker = mocker.patch("MicrosoftTeams.set_integration_context", return_value={})
debug_log_mocker = mocker.patch.object(demisto, "debug")
mocker.patch("MicrosoftTeams.AUTH_TYPE", new="Authorization Code")
auth_type_switch_handling()
assert set_integration_context_mocker.call_count == 1
assert set_integration_context_mocker.call_args[0][0] == {"current_auth_type": "Authorization Code"}
assert "This is the first run of the integration instance" in debug_log_mocker.call_args[0][0]
assert debug_log_mocker.call_count == 1
def test_message_update(mocker, requests_mock):
"""
Given:
- a message as a basic string and a message that contains GUID.
When:
- running send message function.
Then:
- The message is sent successfully in both cases.
"""
from MicrosoftTeams import message_update_command
mocker.patch.object(demisto, "results")
mocker.patch("MicrosoftTeams.get_channel_type", return_value="standard")
expected = util_load_json("test_data/send_message/expected_generic.json")
raw = util_load_json("test_data/send_message/raw_generic.json")
activity_id: str = "1730232813350"
conversation_id: str = "19:2cbad0d78c624400ef83a5750534448g@thread.skype"
mocker.patch("MicrosoftTeams.BOT_ID", new=bot_id)
mocker.patch.object(
demisto,
"args",
return_value={
"message_id": activity_id,
"team": team_name,
"channel": "incident-10",
"message": "Updated message",
"format_as_card": False,
},
)
requests_mock.put(f"{service_url}/v3/conversations/{conversation_id}/activities/{activity_id}", json={"id": activity_id})
requests_mock.post(f"{service_url}/v3/conversations/{mirrored_channels[0]['channel_id']}/activities", json=raw)
message_update_command()
results = demisto.results.call_args[0]
assert len(results) == 1
assert results[0] == expected
@pytest.mark.parametrize(
"permissions, expected_out",
[
(
[Perms.GROUP_READWRITE_ALL.value],
{
Perms.GROUP_READWRITE_ALL,
Perms.GROUP_READ_ALL,
Perms.GROUPMEMBER_READ_ALL,
Perms.CHANNEL_CREATE,
Perms.CHANNEL_READBASIC_ALL,
Perms.CHANNEL_DELETE_ALL,
Perms.CHANNELMESSAGE_SEND,
},
),
(
[Perms.CHAT_READWRITE.value, Perms.USER_READ_ALL.value, "UnknownPerm.Read"],
{
Perms.CHAT_READWRITE,
Perms.CHAT_READ,
Perms.CHAT_READBASIC,
Perms.CHAT_CREATE,
Perms.CHATMESSAGE_SEND,
Perms.USER_READ_ALL,
Perms.USER_READ,
"UnknownPerm.Read",
},
),
],
)
def test_expand_permissions_list(permissions, expected_out):
"""
Given:
- A list of Microsoft Graph permissions.
When:
- The `expand_permission_list` function is called.
Then:
- The permission list is expanded to include relevant related permissions.
"""
from MicrosoftTeams import expand_permission_list
expanded_permissions = expand_permission_list(permissions)
assert expanded_permissions == expected_out
@pytest.mark.parametrize(
"auth_type, command, expected_missing",
[
(CLIENT_CREDENTIALS_FLOW, "microsoft-teams-create-channel", {Perms.CHANNEL_CREATE, Perms.GROUPMEMBER_READ_ALL}),
(
AUTHORIZATION_CODE_FLOW,
"microsoft-teams-message-send-to-chat",
{Perms.CHAT_READBASIC, Perms.CHAT_CREATE, Perms.APPCATALOG_READ_ALL, Perms.TEAMSAPPINSTALLATION_READWRITESELFFORCHAT},
),
],
)
def test_insufficient_permissions_handler(mocker, auth_type, command, expected_missing):
"""
Given:
- Microsoft Graph API returns a 403 Forbidden error due to insufficient permissions.
When:
- The `handle_insufficient_permissions` function is called.
Then:
- Verify that the function determines the missing permissions as expected.
"""
from MicrosoftTeams import insufficient_permissions_error_handler, create_missing_permissions_section
mock_permissions = [Perms.USER_READ_ALL.value, Perms.CHATMESSAGE_SEND.value]
mocker.patch.object(demisto, "command", return_value=command)
mocker.patch("MicrosoftTeams.get_token_permissions", return_value=mock_permissions)
mocker.patch(
"MicrosoftTeams.get_integration_context",
return_value={
CREDENTIALS_TOKEN_PARAMS: json.dumps({"graph_access_token": "mock_token"}),
AUTHCODE_TOKEN_PARAMS: json.dumps({"graph_access_token": "mock_token"}),
},
)
missing_permissions_mock = mocker.patch(
"MicrosoftTeams.create_missing_permissions_section", side_effect=create_missing_permissions_section
)
error_msg = insufficient_permissions_error_handler(auth_type=auth_type)
assert error_msg
assert set(missing_permissions_mock.call_args[0][0]) == expected_missing
def test_commands_required_includes_all_commands():
"""
A list of required permissions should be added to COMMANDS_REQUIRED_PERMISSIONS
whenever a new command is added.
Given:
- COMMANDS_REQUIRED_PERMISSIONS dict.
When:
- An integration command is defined in the yml.
Then:
- A permissions required entry exists in the dict for the command.
"""
from MicrosoftTeams import COMMANDS_REQUIRED_PERMISSIONS
import yaml
try:
with open("MicrosoftTeams.yml") as f:
yml = yaml.safe_load(f)
except FileNotFoundError:
pytest.skip("yml file is unavailable for testing in this environment")
for command in yml["script"]["commands"]:
assert command["name"] in COMMANDS_REQUIRED_PERMISSIONS[AUTHORIZATION_CODE_FLOW]
assert command["name"] in COMMANDS_REQUIRED_PERMISSIONS[CLIENT_CREDENTIALS_FLOW]
def test_get_one_on_one_chat_id(mocker, requests_mock):
"""
Test functionality for getting the chat id of oneOnOne chats without using the chat creation endpoint.
Given:
- The user_id of a oneOnOne chat recipient
When:
- Calling the get_one_on_one_chat_id function
Then:
- Ensure the expected request body is sent and the chat id is retrieved successfully
"""
from MicrosoftTeams import get_one_on_one_chat_id
mock_signed_in_response = test_data.get("signed_in_user")
mocker.patch("MicrosoftTeams.get_signed_in_user", return_value=mock_signed_in_response)
mock_chat_response = test_data.get("get_oneOnOne_chat_id_response")
mock_user_id = mock_chat_response.get("value")[0].get("members")[0].get("userId")
expected_request_qs = {
"$expand": ["members"],
"$filter": [
f"chattype eq 'oneonone' and members/any(m:m/microsoft.graph.aaduserconversationmember/userid eq '{mock_user_id}')"
],
}
expected_chat_id = mock_chat_response.get("value")[0].get("id")
requests_mock.get("https://graph.microsoft.com/v1.0/me/chats", json=mock_chat_response)
chat_id = get_one_on_one_chat_id(mock_user_id)
assert requests_mock.request_history[1].qs == expected_request_qs
assert chat_id == expected_chat_id
def test_get_chat_id_and_type_no_chat_creation_permission(mocker, requests_mock):
"""
Test get_chat_id_and_type function when trying to get a oneOnOne chat and chat creation is forbidden.
Given:
The 'chat' argument as member name
When:
- Calling the get_chat_id_and_type function with create_dm_chat = False
Then:
- There will be no attempt to create the chat
- The chat_id is returned correctly
"""
from MicrosoftTeams import get_chat_id_and_type
mock_chat_id = "test_oneonone_chat_id"
mock_user_id = "user_id"
chat_name = "test_admin"
requests_mock.get(
f"https://graph.microsoft.com/v1.0/chats/?$select=id, chatType&$filter=topic eq '{chat_name}'",
json=test_data.get("get_chat_id_and_type_no_chat_response"),
)
get_user_mock = mocker.patch("MicrosoftTeams.get_user", return_value=[{"id": mock_user_id, "userType": "Member"}])
get_one_on_one_chat_id_mock = mocker.patch("MicrosoftTeams.get_one_on_one_chat_id", return_value=mock_chat_id)
create_chat_mock = mocker.patch("MicrosoftTeams.create_chat")
assert get_chat_id_and_type(chat_name, create_dm_chat=False) == (mock_chat_id, "oneOnOne")
create_chat_mock.assert_not_called()
get_one_on_one_chat_id_mock.assert_called_once_with(mock_user_id)
assert get_user_mock.call_count == 1
def test_get_chat_id_and_type_not_found_no_chat_creation_permission(mocker, requests_mock):
"""
Test get_chat_id_and_type function when trying to get a non-existent oneOnOne chat and chat creation is forbidden.
Given:
- chat as a user_id that does not have an existing oneOnOne chat
When:
- Calling the get_chat_id_and_type function with create_dm_chat = False
Then:
- A chat not found error is raised
"""
from MicrosoftTeams import get_chat_id_and_type
mock_signed_in_response = test_data.get("signed_in_user")
mocker.patch("MicrosoftTeams.get_signed_in_user", return_value=mock_signed_in_response)
mock_chat_response = test_data.get("get_oneOnOne_chat_id_no_chat_response")
chat_name = "unknown"
requests_mock.get(
f"https://graph.microsoft.com/v1.0/chats/?$select=id, chatType&$filter=topic eq '{chat_name}'",
json=test_data.get("get_chat_id_and_type_no_chat_response"),
)
requests_mock.get("https://graph.microsoft.com/v1.0/me/chats", json=mock_chat_response)
mocker.patch("MicrosoftTeams.get_user", return_value=[{"id": "mock_user_id", "userType": "Member"}])
with pytest.raises(ValueError) as e:
get_chat_id_and_type(chat_name, create_dm_chat=False)
assert str(e.value) == f"Could not find chat: {chat_name}"
def test_ring_user_in_auth_code(mocker, requests_mock):
"""
Given:
- Authentication type is set to authorization code.
When:
- Calling the ring-user command.
Then:
- Ring user is called with a token obtained from client credentials instead of auth code.
"""
from MicrosoftTeams import ring_user
mocker.patch("MicrosoftTeams.AUTH_TYPE", new=AUTHORIZATION_CODE_FLOW)
mocker.patch.object(demisto, "args", return_value={"username": "test_user"})
mocker.patch("MicrosoftTeams.get_user", return_value=[{"id": "test_user_id", "userType": "Member"}])
mock_credentials_token = "credentials_token"
requests_mock.post(
f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
json={"access_token": mock_credentials_token},
status_code=200,
additional_matcher=lambda request: "grant_type=client_credentials" in request.text,
)
requests_mock.post(
f"{GRAPH_BASE_URL}/v1.0/communications/calls",
json={},
)
integration_context.pop(CREDENTIALS_TOKEN_PARAMS)
ring_user()
assert requests_mock.last_request.headers["Authorization"] == f"Bearer {mock_credentials_token}"
@freeze_time("2025-03-20")
@pytest.mark.parametrize("auth_type", [AUTHORIZATION_CODE_FLOW, CLIENT_CREDENTIALS_FLOW])
def test_integration_context_format_migration(mocker, auth_type):
"""
Test the context migration logic from the old single graph token format to
handling both token types individually.
Given:
- The integration context contains a cached graph token in the old format.
When:
- Requesting a graph access token.
Then:
- The integration context is updated to the new format.
- The existing token is migrated to the appropriate token type.
- The existing token is returned as expected.
"""
from MicrosoftTeams import get_graph_access_token, AUTHCODE_TOKEN_PARAMS, CREDENTIALS_TOKEN_PARAMS
mock_integration_context = {
"current_refresh_token": "mock_refresh_token",
"graph_access_token": "mock_cached_token",
"graph_valid_until": 1742438800,
}
mocker.patch.object(demisto, "getIntegrationContext", return_value=mock_integration_context)
mocker.patch("MicrosoftTeams.AUTH_TYPE", new=auth_type)
token = get_graph_access_token()
assert token == "mock_cached_token"
context_token_key = AUTHCODE_TOKEN_PARAMS if auth_type == AUTHORIZATION_CODE_FLOW else CREDENTIALS_TOKEN_PARAMS
assert json.loads(str(mock_integration_context.get(context_token_key, "{}"))) == {
"current_refresh_token": "mock_refresh_token",
"graph_access_token": "mock_cached_token",
"graph_valid_until": 1742438800,
}
assert "current_refresh_token" not in mock_integration_context
assert "graph_access_token" not in mock_integration_context
assert "graph_valid_until" not in mock_integration_context
@freeze_time("2025-03-20")
def test_integration_context_format_migration_auth_override(mocker, requests_mock):
"""
Test the context migration logic when the requested auth type is different from the integration configuration.
Given:
- The integration context contains a cached graph token in the old format.
When:
- Requesting a graph access token.
- The request is made with the auth type argument overriding the integration configuration.
Then:
- The integration context is updated to the new format.
- The existing token is migrated to the correct token type key.
- A token of the requested type is created and returned as expected.
"""
from MicrosoftTeams import get_graph_access_token, CREDENTIALS_TOKEN_PARAMS, AUTHCODE_TOKEN_PARAMS
mock_integration_context = {
"tenant_id": tenant_id,
"current_refresh_token": "mock_refresh_token",
"graph_access_token": "mock_cached_token",
"graph_valid_until": 1742438800,
}
mocker.patch.object(demisto, "getIntegrationContext", return_value=mock_integration_context)
mocker.patch("MicrosoftTeams.AUTH_TYPE", new=CLIENT_CREDENTIALS_FLOW)
mocker.patch("MicrosoftTeams.AUTH_CODE", new="mock_auth_code")
requests_mock.post(
f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
json={
"access_token": "mock_auth_token",
"refresh_token": "mock_auth_refresh_token",
"expires_in": 3600,
},
)
token = get_graph_access_token(AUTHORIZATION_CODE_FLOW)
assert token == "mock_auth_token"
assert json.loads(str(mock_integration_context.get(CREDENTIALS_TOKEN_PARAMS, "{}"))) == {
"current_refresh_token": "mock_refresh_token",
"graph_access_token": "mock_cached_token",
"graph_valid_until": 1742438800,
}
assert json.loads(str(mock_integration_context.get(AUTHCODE_TOKEN_PARAMS, "{}"))) == {
"current_refresh_token": "mock_auth_refresh_token",
"graph_access_token": "mock_auth_token",
"graph_valid_until": 1742432395,
}
assert get_graph_access_token() == "mock_cached_token"
def test_team_deleted_message_handler(mocker):
"""
Given:
- Integration context contains team entries with some name.
When:
- An event message is received about that team's name changing.
Then:
- The team name is changed in the integration cache.
"""
from MicrosoftTeams import APP
mock_request_body = {
"channelData": {
"eventType": "teamDeleted",
"team": {
"aadGroupId": "33333333-3333-3333-3333-333333333333",
"id": "19:333333333333333333333333333333333333333333333333333333333",
"name": "Test3",
},
"tenant": {
"id": "00000000-0000-0000-0000-000000000000",
},
},
"channelId": "msteams",
"conversation": {
"conversationType": "channel",
"id": "19:000000000000000000000000000000000000000000000000000000000",
"isGroup": True,
"tenantId": "00000000-0000-0000-0000-000000000000",
},
"from": {
"aadObjectId": "00000000-0000-0000-0000-000000000000",
"id": "29:000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
},
"id": "f:00000000-0000-0000-0000-000000000000",
"recipient": {
"id": "28:000000000000000000000000000000000000",
"name": "TestBot",
},
"type": "conversationUpdate",
}
mock_teams_cache = [
{
"team_aad_id": "11111111-1111-1111-1111-111111111111",
"team_id": "19:111111111111111111111111111111111111111111111111111111111",
"team_members": [],
"team_name": "Test",
},
{
"team_aad_id": "22222222-2222-2222-2222-222222222222",
"team_id": "19:222222222222222222222222222222222222222222222222222222222",
"team_members": [],
"team_name": "Test2",
},
{
"team_aad_id": "33333333-3333-3333-3333-333333333333",
"team_id": "19:333333333333333333333333333333333333333333333333333333333",
"team_members": [],
"team_name": "Test3",
},
]
mocker.patch("MicrosoftTeams.get_integration_context", return_value={"teams": json.dumps(mock_teams_cache)})
set_context_mock = mocker.patch("MicrosoftTeams.set_integration_context")
mocker.patch("MicrosoftTeams.validate_auth_header", return_value=True)
APP.testing = True
app = APP.test_client()
app.post("/", data=json.dumps(mock_request_body), content_type="application/json")
set_context_mock.assert_called_once_with({"teams": json.dumps(mock_teams_cache[:2])})
def test_team_renamed_message_handler(mocker):
"""
Given:
- Integration context contains team entries with some name.
When:
- An event message is received about that team's name changing.
Then:
- The team name is changed in the integration cache.
"""
from MicrosoftTeams import APP
mock_request_body = {
"channelData": {
"eventType": "teamRenamed",
"team": {
"aadGroupId": "33333333-3333-3333-3333-333333333333",
"id": "19:333333333333333333333333333333333333333333333333333333333",
"name": "Test30",
},
"tenant": {
"id": "00000000-0000-0000-0000-000000000000",
},
},
"channelId": "msteams",
"conversation": {
"conversationType": "channel",
"id": "19:000000000000000000000000000000000000000000000000000000000",
"isGroup": True,
"tenantId": "00000000-0000-0000-0000-000000000000",
},
"from": {
"aadObjectId": "00000000-0000-0000-0000-000000000000",
"id": "29:000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
},
"id": "f:00000000-0000-0000-0000-000000000000",
"recipient": {
"id": "28:000000000000000000000000000000000000",
"name": "TestBot",
},
"type": "conversationUpdate",
}
mock_teams_cache = [
{
"team_aad_id": "11111111-1111-1111-1111-111111111111",
"team_id": "19:111111111111111111111111111111111111111111111111111111111",
"team_members": [],
"team_name": "Test",
},
{
"team_aad_id": "22222222-2222-2222-2222-222222222222",
"team_id": "19:222222222222222222222222222222222222222222222222222222222",
"team_members": [],
"team_name": "Test2",
},
{
"team_aad_id": "33333333-3333-3333-3333-333333333333",
"team_id": "19:333333333333333333333333333333333333333333333333333333333",
"team_members": [],
"team_name": "Test3",
},
]
mocker.patch("MicrosoftTeams.get_integration_context", return_value={"teams": json.dumps(mock_teams_cache)})
set_context_mock = mocker.patch("MicrosoftTeams.set_integration_context")
mocker.patch("MicrosoftTeams.validate_auth_header", return_value=True)
APP.testing = True
app = APP.test_client()
app.post("/", data=json.dumps(mock_request_body), content_type="application/json")
mock_teams_cache[2]["team_name"] = "Test30"
set_context_mock.assert_called_once_with({"teams": json.dumps(mock_teams_cache)})
def test_message_handler_filters_invalid_cache_entries(mocker, requests_mock):
"""
Given:
- Integration context contains multiple team entries with the same name.
When:
- An event message is received about a team with that name.
- One of the cached teams no longer exists.
Then:
- The validity of each entry is tested and invalid entries are removed.
"""
from MicrosoftTeams import APP
mock_request_body = {
"channelData": {
"eventType": "teamRenamed",
"team": {
"aadGroupId": "33333333-3333-3333-3333-333333333333",
"id": "19:333333333333333333333333333333333333333333333333333333333",
"name": "Test2",
},
"tenant": {
"id": "00000000-0000-0000-0000-000000000000",
},
},
"channelId": "msteams",
"conversation": {
"conversationType": "channel",
"id": "19:000000000000000000000000000000000000000000000000000000000",
"isGroup": True,
"tenantId": "00000000-0000-0000-0000-000000000000",
},
"from": {
"aadObjectId": "00000000-0000-0000-0000-000000000000",
"id": "29:000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
},
"id": "f:00000000-0000-0000-0000-000000000000",
"recipient": {
"id": "28:000000000000000000000000000000000000",
"name": "TestBot",
},
"type": "conversationUpdate",
}
mock_teams_cache = [
{
"team_aad_id": "11111111-1111-1111-1111-111111111111",
"team_id": "19:111111111111111111111111111111111111111111111111111111111",
"team_members": [],
"team_name": "Test",
},
{
"team_aad_id": "22222222-2222-2222-2222-222222222222",
"team_id": "19:222222222222222222222222222222222222222222222222222222222",
"team_members": [],
"team_name": "Test2",
},
{
"team_aad_id": "33333333-3333-3333-3333-333333333333",
"team_id": "19:333333333333333333333333333333333333333333333333333333333",
"team_members": [],
"team_name": "Test3",
},
]
mock_team_query_response = {
"value": [
{
"displayName": "Test2",
"id": "33333333-3333-3333-3333-333333333333",
},
]
}
url = f"{GRAPH_BASE_URL}/v1.0/groups?$filter=displayName eq 'Test2' and resourceProvisioningOptions/Any(x:x eq 'Team')"
mocker.patch("MicrosoftTeams.get_integration_context", return_value={"teams": json.dumps(mock_teams_cache)})
mocker.patch("MicrosoftTeams.get_graph_access_token", return_value="mock_token")
set_context_mock = mocker.patch("MicrosoftTeams.set_integration_context")
mocker.patch("MicrosoftTeams.validate_auth_header", return_value=True)
requests_mock.get(url, json=mock_team_query_response)
APP.testing = True
app = APP.test_client()
app.post("/", data=json.dumps(mock_request_body), content_type="application/json")
# Check that the old invalid Test2 was removed, and "Test3" name was changed to "Test2"
mock_teams_cache[2]["team_name"] = "Test2"
mock_teams_cache.pop(1)
set_context_mock.assert_called_once_with({"teams": json.dumps(mock_teams_cache)})
def test_send_notification_with_adaptive_card_from_DlpAskFeedback(mocker, requests_mock):
"""
Given:
- An adaptive_card argument send form DlpAskFeedback.
When:
- Call send_message method.
Then:
- The request to the endpoint is sent with the adaptive card.
"""
from MicrosoftTeams import send_message
team = "The-A-Team"
adaptive_card: dict = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": [{"type": "TextBlock", "text": "message", "wrap": True}],
"actions": [
{
"type": "Action.Submit",
"title": "Yes",
"data": {
"response": "Yes",
"entitlement": "4111dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "1",
"task_id": "1",
},
},
{
"type": "Action.Submit",
"title": "No",
"data": {
"response": "No",
"entitlement": "4111dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "1",
"task_id": "1",
},
},
],
},
}
requests_mock.post(f"{service_url}/v3/conversations", json={"id": "conversation-id"})
mocker.patch("MicrosoftTeams.get_team_aad_id", return_value=team_aad_id)
mocker.patch.object(demisto, "params", return_value={"team": team})
mocker.patch.object(demisto, "args", return_value={"adaptive_card": json.dumps(adaptive_card), "to": "bwillis@email.com"})
send_message_request = requests_mock.post(f"{service_url}/v3/conversations/conversation-id/activities", json={})
send_message()
assert send_message_request.last_request.json() == {"type": "message", "attachments": [adaptive_card]}
def test_send_notification_with_adaptive_card_from_TeamAsk(mocker, requests_mock):
"""
Given:
- An adaptive_card argument send form MicrosoftTeamsAsk.
When:
- Call send_message method.
Then:
- The request to the endpoint is sent with the adaptive card.
"""
from MicrosoftTeams import send_message
adaptive_card: dict = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": [{"type": "TextBlock", "text": "message", "wrap": True}],
"actions": [
{"type": "Action.Submit", "title": "Yes"},
{"type": "Action.Submit", "title": "No"},
],
},
}
adaptive_card_arg_from_TeamsAsk: dict = {
"adaptive_card": adaptive_card,
"entitlement": "4111dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "1",
"task_id": "1",
}
process_adaptive_card = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": [{"type": "TextBlock", "text": "message", "wrap": True}],
"actions": [
{
"type": "Action.Submit",
"title": "Yes",
"data": {"entitlement": "4111dae8-2d45-46bd-85fa-64779c12abe8", "investigation_id": "1", "task_id": "1"},
},
{
"type": "Action.Submit",
"title": "No",
"data": {"entitlement": "4111dae8-2d45-46bd-85fa-64779c12abe8", "investigation_id": "1", "task_id": "1"},
},
],
},
}
requests_mock.post(f"{service_url}/v3/conversations", json={"id": "conversation-id"})
mocker.patch("MicrosoftTeams.get_team_aad_id", return_value=team_aad_id)
mocker.patch.object(demisto, "params", return_value={"team": "The-A-Team"})
mocker.patch.object(
demisto, "args", return_value={"adaptive_card": json.dumps(adaptive_card_arg_from_TeamsAsk), "to": "bwillis@email.com"}
)
send_message_request = requests_mock.post(f"{service_url}/v3/conversations/conversation-id/activities", json={})
send_message()
assert send_message_request.last_request.json() == {"type": "message", "attachments": [process_adaptive_card]}
def test_process_ask_user():
"""
Given:
- A message object send form MicrosoftTeamsAsk.
When:
- Call process_ask_user method.
Then:
- The adaptive card was created as expected.
"""
from MicrosoftTeams import process_ask_user
import json
message = {
"message_text": "message_text",
"options": ["Yes", "No"],
"entitlement": "4111dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "1",
"task_id": "1",
"form_type": "predefined-options",
}
expected_adaptive_card = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": [{"type": "TextBlock", "text": "message_text", "wrap": True}],
"actions": [
{
"type": "Action.Submit",
"title": "Yes",
"data": {
"response": "Yes",
"entitlement": "4111dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "1",
"task_id": "1",
},
},
{
"type": "Action.Submit",
"title": "No",
"data": {
"response": "No",
"entitlement": "4111dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "1",
"task_id": "1",
},
},
],
},
}
assert process_ask_user(json.dumps(message)) == expected_adaptive_card
def test_send_notification_with_raw_adaptive_card(mocker, requests_mock):
"""
Given:
- A raw adaptive_card input.
When:
- send-notification is called.
Then:
- The adaptive card is wrapped with the required contentType and content fields.
- The request to the endpoint is sent with the adaptive card as expected.
"""
from MicrosoftTeams import send_message
adaptive_card: dict = {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": [{"type": "TextBlock", "text": "message", "wrap": True}],
"actions": [
{"type": "Action.Submit", "title": "Yes"},
{"type": "Action.Submit", "title": "No"},
],
}
expected_request_attachment = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": adaptive_card,
}
requests_mock.post(f"{service_url}/v3/conversations", json={"id": "conversation-id"})
mocker.patch("MicrosoftTeams.get_team_aad_id", return_value=team_aad_id)
mocker.patch.object(demisto, "params", return_value={"team": "The-A-Team"})
mocker.patch.object(demisto, "args", return_value={"adaptive_card": json.dumps(adaptive_card), "to": "bwillis@email.com"})
send_message_request = requests_mock.post(f"{service_url}/v3/conversations/conversation-id/activities", json={})
send_message()
assert send_message_request.last_request.json() == {"type": "message", "attachments": [expected_request_attachment]}
def test_send_notification_with_raw_adaptive_card_from_TeamAsk(mocker, requests_mock):
"""
Given:
- A raw adaptive_card is sent from MicrosoftTeamsAsk.
When:
- Call send_message method.
Then:
- The adaptive card is wrapped with the required contentType and content fields.
- The request to the endpoint is sent with the adaptive card.
"""
from MicrosoftTeams import send_message
adaptive_card: dict = {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": [{"type": "TextBlock", "text": "message", "wrap": True}],
"actions": [
{"type": "Action.Submit", "title": "Yes"},
{"type": "Action.Submit", "title": "No"},
],
}
adaptive_card_arg_from_TeamsAsk: dict = {
"adaptive_card": adaptive_card,
"entitlement": "4111dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "1",
"task_id": "1",
}
expected_request_attachment = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.0",
"type": "AdaptiveCard",
"msteams": {"width": "Full"},
"body": [{"type": "TextBlock", "text": "message", "wrap": True}],
"actions": [
{
"type": "Action.Submit",
"title": "Yes",
"data": {"entitlement": "4111dae8-2d45-46bd-85fa-64779c12abe8", "investigation_id": "1", "task_id": "1"},
},
{
"type": "Action.Submit",
"title": "No",
"data": {"entitlement": "4111dae8-2d45-46bd-85fa-64779c12abe8", "investigation_id": "1", "task_id": "1"},
},
],
},
}
requests_mock.post(f"{service_url}/v3/conversations", json={"id": "conversation-id"})
mocker.patch("MicrosoftTeams.get_team_aad_id", return_value=team_aad_id)
mocker.patch.object(demisto, "params", return_value={"team": "The-A-Team"})
mocker.patch.object(
demisto, "args", return_value={"adaptive_card": json.dumps(adaptive_card_arg_from_TeamsAsk), "to": "bwillis@email.com"}
)
send_message_request = requests_mock.post(f"{service_url}/v3/conversations/conversation-id/activities", json={})
send_message()
assert send_message_request.last_request.json() == {"type": "message", "attachments": [expected_request_attachment]}
def test_get_bot_access_token_multi_tenant_success(mocker, requests_mock):
"""
Given:
- A multi-tenant bot configuration.
When:
- Calling get_bot_access_token.
Then:
- Ensure a token is successfully retrieved using the multi-tenant endpoint.
"""
from MicrosoftTeams import get_bot_access_token
mocker.patch.object(demisto, "getIntegrationContext", return_value={})
mocker.patch.object(demisto, "setIntegrationContext")
requests_mock.post(
"https://login.microsoftonline.com/botframework.com/oauth2/v2.0/token",
json={"access_token": "multi_tenant_token", "expires_in": 3600},
)
token = get_bot_access_token()
assert token == "multi_tenant_token"
def test_get_bot_access_token_single_tenant_success(mocker, requests_mock):
"""
Given:
- A single-tenant bot configuration with a tenant_id.
When:
- Calling get_bot_access_token.
Then:
- Ensure a token is successfully retrieved using the single-tenant endpoint.
"""
from MicrosoftTeams import get_bot_access_token
mocker.patch.object(demisto, "getIntegrationContext", return_value={"bot_type": "single-tenant", "tenant_id": tenant_id})
mocker.patch.object(demisto, "setIntegrationContext")
requests_mock.post(
f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
json={"access_token": "single_tenant_token", "expires_in": 3600},
)
token = get_bot_access_token()
assert token == "single_tenant_token"
def test_get_bot_access_token_fallback_to_single_tenant(mocker, requests_mock):
"""
Given:
- A multi-tenant bot configuration.
- The multi-tenant endpoint returns an 'unauthorized_client' error.
- A tenant_id is available in the context.
When:
- Calling get_bot_access_token.
Then:
- Ensure the code falls back to the single-tenant endpoint and retrieves a token.
"""
from MicrosoftTeams import get_bot_access_token
mocker.patch.object(demisto, "getIntegrationContext", return_value={"tenant_id": tenant_id})
set_context_mocker = mocker.patch.object(demisto, "setIntegrationContext")
requests_mock.post(
"https://login.microsoftonline.com/botframework.com/oauth2/v2.0/token",
json={"error": "unauthorized_client"},
status_code=400,
)
requests_mock.post(
f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
json={"access_token": "fallback_token", "expires_in": 3600},
)
token = get_bot_access_token()
assert token == "fallback_token"
# Verify that the bot_type was updated in the context for future use
updated_context = set_context_mocker.call_args[0][0]
assert updated_context.get("bot_type") == "single-tenant"
def test_get_bot_access_token_single_tenant_no_tenant_id(mocker):
"""
Given:
- A single-tenant bot configuration but no tenant_id.
When:
- Calling get_bot_access_token.
Then:
- Ensure a ValueError is raised.
"""
from MicrosoftTeams import get_bot_access_token, MISS_CONFIGURATION_ERROR_MESSAGE
mocker.patch.object(demisto, "getIntegrationContext", return_value={"bot_type": "single-tenant"})
mocker.patch.object(demisto, "setIntegrationContext")
with pytest.raises(ValueError, match=MISS_CONFIGURATION_ERROR_MESSAGE):
get_bot_access_token()
def test_validate_auth_header_signature_verification(mocker):
"""
Given:
- A valid JWT token signed with a private key.
- An invalid JWT token signed with a different private key (attacker's key).
- The public key corresponding to the valid token is available in the integration context (JWK).
When:
- Calling validate_auth_header with the valid token.
- Calling validate_auth_header with the invalid token.
Then:
- The valid token should be accepted (return True).
- The invalid token should be rejected (return False) because signature verification fails.
"""
from MicrosoftTeams import validate_auth_header
import jwt
import json
import base64
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.backends import default_backend
# 1. Setup Keys
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend())
public_key = private_key.public_key()
attacker_private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend())
def base64_url_encode(val):
bytes_val = val.to_bytes((val.bit_length() + 7) // 8, byteorder="big")
return base64.urlsafe_b64encode(bytes_val).decode("utf-8").rstrip("=")
def get_jwk(pub_key, kid):
numbers = pub_key.public_numbers()
return {
"kty": "RSA",
"kid": kid,
"n": base64_url_encode(numbers.n),
"e": base64_url_encode(numbers.e),
"alg": "RS256",
"use": "sig",
"endorsements": ["msteams"],
}
kid = "test-key-id"
jwk = get_jwk(public_key, kid)
# 2. Mock Integration Context
mocker.patch("MicrosoftTeams.get_integration_context", return_value={"open_id_metadata": json.dumps({"keys": [jwk]})})
mocker.patch("MicrosoftTeams.set_integration_context")
mocker.patch("MicrosoftTeams.BOT_ID", new="test-bot-id")
# 3. Create Tokens
payload = {"iss": "https://api.botframework.com", "aud": "test-bot-id", "sub": "test-user", "exp": 9999999999}
valid_token = jwt.encode(payload=payload, key=private_key, algorithm="RS256", headers={"kid": kid})
invalid_token = jwt.encode(payload=payload, key=attacker_private_key, algorithm="RS256", headers={"kid": kid})
# 4. Run Tests
# Test Valid Token
headers_valid = {"Authorization": f"Bearer {valid_token}"}
is_valid = validate_auth_header(headers_valid)
assert is_valid is True
# Test Invalid Token
headers_invalid = {"Authorization": f"Bearer {invalid_token}"}
is_valid = validate_auth_header(headers_invalid)
assert is_valid is False
def test_messages_endpoint_auth_header_validation_failure(mocker):
"""
Given:
- A POST request to the messages endpoint with invalid authorization headers.
When:
- The validate_auth_header function returns False.
Then:
- The endpoint returns a 401 status code.
- The response contains a generic error message.
"""
from MicrosoftTeams import APP
# Mock validate_auth_header to return False
mocker.patch("MicrosoftTeams.validate_auth_header", return_value=False)
# Mock demisto.info to avoid errors
mocker.patch.object(demisto, "info")
# Create test request with invalid headers
mock_request_body = {
"type": "message",
"text": "test message",
"from": {"id": "test-user-id", "name": "Test User"},
"conversation": {"id": "test-conversation-id"},
}
APP.testing = True
app = APP.test_client()
# Send POST request with invalid authorization
response = app.post(
"/",
data=json.dumps(mock_request_body),
content_type="application/json",
headers={"Authorization": "Bearer invalid_token"},
)
# Assert response
assert response.status_code == 401
assert response.data.decode() == "Authorization header validation failed - JWT validation error"
@pytest.mark.parametrize(
"args, expected_response, expected_request_url, expected_outputs",
[
(
{"conversation_id": GROUP_CHAT_ID, "limit": 2, "order_by": "createdDateTime"},
"list_messages",
f"https://graph.microsoft.com/v1.0/chats/{GROUP_CHAT_ID}/messages?$top=2&$orderBy=createdDateTime desc",
"expected_outputs_list_messages_new_command",
),
(
{
"next_link": "https://graph.microsoft.com/v1.0/chats/test_next_link",
"limit": 2,
"conversation_id": GROUP_CHAT_ID,
},
"list_messages",
"https://graph.microsoft.com/v1.0/chats/test_next_link",
"expected_outputs_list_messages_new_command",
),
],
)
def test_list_messages_command_chat(mocker, requests_mock, args, expected_response, expected_request_url, expected_outputs):
"""
Given:
- The command arguments for listing messages in a chat
When:
- Executing the 'microsoft-teams-list-messages' command.
Then:
- Assert the request url is as expected
- Verify that the context outputs is as expected
"""
from MicrosoftTeams import list_messages_command
mocker.patch("MicrosoftTeams.AUTH_TYPE", new=AUTHORIZATION_CODE_FLOW)
mocker.patch("MicrosoftTeams.get_chat_id_and_type", return_value=(GROUP_CHAT_ID, "group"))
return_results = mocker.patch("MicrosoftTeams.return_results")
mocker.patch.object(demisto, "args", return_value=args)
requests_mock.get(expected_request_url, json=test_data.get(expected_response))
list_messages_command()
assert return_results.call_args[0][0].outputs == test_data.get(expected_outputs)
def test_list_messages_command_channel(mocker, requests_mock):
"""
Given:
- The command arguments for listing messages in a channel
When:
- Executing the 'microsoft-teams-list-messages' command.
Then:
- Assert the request url is as expected
- Verify that the context outputs is as expected
"""
from MicrosoftTeams import list_messages_command
channel_name = "incident-1"
team_name = "The-A-Team"
channel_id = mirrored_channels[0]["channel_id"]
args = {"conversation_id": channel_name, "team_name": team_name, "limit": 2}
mocker.patch("MicrosoftTeams.get_team_aad_id", return_value=team_aad_id)
mocker.patch("MicrosoftTeams.get_channel_id", return_value=channel_id)
return_results = mocker.patch("MicrosoftTeams.return_results")
mocker.patch.object(demisto, "args", return_value=args)
expected_request_url = f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels/{channel_id}/messages?$top=2"
requests_mock.get(expected_request_url, json=test_data.get("list_messages"))
list_messages_command()
expected_outputs = test_data.get("expected_outputs_list_messages_new_command")
# Adjust expected outputs for channel context
expected_outputs["MicrosoftTeams.MessagesList(val.conversationId && val.conversationId === obj.conversationId)"][
"conversationId"
] = channel_name
assert return_results.call_args[0][0].outputs == expected_outputs
def test_list_messages_command_replies(mocker, requests_mock):
"""
Given:
- The command arguments for listing replies to a message in a channel
When:
- Executing the 'microsoft-teams-list-messages' command.
Then:
- Assert the request url is as expected
- Verify that the context outputs is as expected
"""
from MicrosoftTeams import list_messages_command
channel_name = "incident-1"
team_name = "The-A-Team"
channel_id = mirrored_channels[0]["channel_id"]
message_id = "1616964509832"
args = {"conversation_id": channel_name, "team_name": team_name, "message_id": message_id, "limit": 2}
mocker.patch("MicrosoftTeams.get_team_aad_id", return_value=team_aad_id)
mocker.patch("MicrosoftTeams.get_channel_id", return_value=channel_id)
return_results = mocker.patch("MicrosoftTeams.return_results")
mocker.patch.object(demisto, "args", return_value=args)
expected_request_url = (
f"https://graph.microsoft.com/v1.0/teams/{team_aad_id}/channels/{channel_id}/messages/{message_id}/replies?$top=2"
)
requests_mock.get(expected_request_url, json=test_data.get("list_messages"))
list_messages_command()
expected_outputs = test_data.get("expected_outputs_list_messages_new_command")
# Adjust expected outputs for channel context
expected_outputs["MicrosoftTeams.MessagesList(val.conversationId && val.conversationId === obj.conversationId)"][
"conversationId"
] = channel_name
assert return_results.call_args[0][0].outputs == expected_outputs
def test_list_messages_command_error(mocker):
"""
Given:
- The command arguments for listing messages in a channel without providing team
When:
- Executing the 'microsoft-teams-list-messages' command.
Then:
- Assert that ValueError is raised
"""
from MicrosoftTeams import list_messages_command
mocker.patch("MicrosoftTeams.AUTH_TYPE", new=AUTHORIZATION_CODE_FLOW)
channel_name = "incident-1"
args = {"conversation_id": channel_name, "limit": 2}
# Mock get_chat_id_and_type to raise exception as it's not a chat
mocker.patch("MicrosoftTeams.get_chat_id_and_type", side_effect=ValueError("Could not find chat"))
mocker.patch.object(demisto, "args", return_value=args)
with pytest.raises(DemistoException) as e:
list_messages_command()
assert (
str(e.value)
== "Failed to find chat or channel. If you are trying to get messages from a channel, please provide the 'team_name'."
)
def test_send_proactive_message_command_with_text_message(mocker, requests_mock):
"""
Given:
- A user identifier and a text message
When:
- Executing the 'microsoft-teams-send-proactive-message' command
Then:
- Verify user is resolved correctly
- Verify conversation is created
- Verify message is sent successfully
- Verify outputs are correct
"""
from MicrosoftTeams import send_proactive_message_command
user_email = "test.user@example.com"
user_id = "test-user-id-12345"
conversation_id = "19:test-conversation-id"
activity_id = "1730232813350"
# Mock get_user to resolve user identifier
requests_mock.get(
f"{GRAPH_BASE_URL}/v1.0/users",
json={"value": [{"id": user_id, "mail": user_email, "displayName": "Test User"}]},
)
# Mock create conversation
requests_mock.post(f"{service_url}/v3/conversations", json={"id": conversation_id})
# Mock send message
requests_mock.post(f"{service_url}/v3/conversations/{conversation_id}/activities", json={"id": activity_id})
mocker.patch.object(demisto, "args", return_value={"user_id": user_email, "message": "Hello from XSOAR!"})
return_results = mocker.patch("MicrosoftTeams.return_results")
send_proactive_message_command()
# Verify outputs
results = return_results.call_args[0][0]
assert results.outputs["ConversationId"] == conversation_id
assert results.outputs["UserId"] == user_id
assert results.outputs["ActivityId"] == activity_id
assert results.outputs["UserIdentifier"] == user_email
assert results.readable_output == "Message was sent successfully."
def test_send_proactive_message_command_with_adaptive_card(mocker, requests_mock):
"""
Given:
- A user identifier and an adaptive card (as dict)
When:
- Executing the 'microsoft-teams-send-proactive-message' command
Then:
- Verify adaptive card is sent correctly
- Verify raw adaptive card is wrapped properly
"""
from MicrosoftTeams import send_proactive_message_command
user_id = "test-user-id-12345"
conversation_id = "19:test-conversation-id"
activity_id = "1730232813351"
adaptive_card = {
"type": "AdaptiveCard",
"version": "1.0",
"body": [{"type": "TextBlock", "text": "Test adaptive card"}],
}
# Mock get_user
requests_mock.get(f"{GRAPH_BASE_URL}/v1.0/users", json={"value": [{"id": user_id, "mail": "test@example.com"}]})
# Mock create conversation
requests_mock.post(f"{service_url}/v3/conversations", json={"id": conversation_id})
# Mock send message
send_message_mock = requests_mock.post(
f"{service_url}/v3/conversations/{conversation_id}/activities", json={"id": activity_id}
)
mocker.patch.object(demisto, "args", return_value={"user_id": "test@example.com", "adaptive_card": json.dumps(adaptive_card)})
send_proactive_message_command()
# Verify adaptive card was wrapped correctly
sent_payload = send_message_mock.last_request.json()
assert sent_payload["type"] == "message"
assert len(sent_payload["attachments"]) == 1
assert sent_payload["attachments"][0]["contentType"] == "application/vnd.microsoft.card.adaptive"
assert sent_payload["attachments"][0]["content"]["type"] == "AdaptiveCard"
def test_send_proactive_message_command_with_teams_ask_adaptive_card(mocker, requests_mock):
"""
Given:
- A user identifier and an adaptive card with entitlement (from MicrosoftTeamsAsk)
When:
- Executing the 'microsoft-teams-send-proactive-message' command
Then:
- Verify entitlement is injected into all Action.Submit elements
- Verify message is sent correctly
"""
from MicrosoftTeams import send_proactive_message_command
user_id = "test-user-id-12345"
conversation_id = "19:test-conversation-id"
adaptive_card_with_entitlement = {
"adaptive_card": {
"type": "AdaptiveCard",
"version": "1.0",
"body": [{"type": "TextBlock", "text": "Approve this request?"}],
"actions": [
{"type": "Action.Submit", "title": "Approve"},
{"type": "Action.Submit", "title": "Deny"},
],
},
"entitlement": "4404dae8-2d45-46bd-85fa-64779c12abe8",
"investigation_id": "100",
"task_id": "5",
}
# Mock get_user
requests_mock.get(f"{GRAPH_BASE_URL}/v1.0/users", json={"value": [{"id": user_id, "mail": "test@example.com"}]})
# Mock create conversation
requests_mock.post(f"{service_url}/v3/conversations", json={"id": conversation_id})
# Mock send message
send_message_mock = requests_mock.post(
f"{service_url}/v3/conversations/{conversation_id}/activities", json={"id": "activity-id"}
)
mocker.patch.object(
demisto, "args", return_value={"user_id": "test@example.com", "adaptive_card": json.dumps(adaptive_card_with_entitlement)}
)
mocker.patch("MicrosoftTeams.return_results")
send_proactive_message_command()
# Verify entitlement was injected
sent_payload = send_message_mock.last_request.json()
actions = sent_payload["attachments"][0]["content"]["actions"]
assert actions[0]["data"]["entitlement"] == "4404dae8-2d45-46bd-85fa-64779c12abe8"
assert actions[0]["data"]["investigation_id"] == "100"
assert actions[0]["data"]["task_id"] == "5"
assert actions[1]["data"]["entitlement"] == "4404dae8-2d45-46bd-85fa-64779c12abe8"
def test_send_proactive_message_command_user_not_found(mocker, requests_mock):
"""
Given:
- A user identifier that doesn't exist
When:
- Executing the 'microsoft-teams-send-proactive-message' command
Then:
- Verify appropriate error is raised
"""
from MicrosoftTeams import send_proactive_message_command
# Mock get_user to return empty list
requests_mock.get(f"{GRAPH_BASE_URL}/v1.0/users", json={"value": []})
mocker.patch.object(demisto, "args", return_value={"user_id": "nonexistent@example.com", "message": "Test"})
# Verify ValueError is raised
with pytest.raises(ValueError) as exc_info:
send_proactive_message_command()
error_msg = str(exc_info.value)
assert "User not found" in error_msg
assert "nonexistent@example.com" in error_msg
def test_send_proactive_message_command_multiple_users_found(mocker, requests_mock):
"""
Given:
- A user identifier that matches multiple users
When:
- Executing the 'microsoft-teams-send-proactive-message' command
Then:
- Verify security error is raised to prevent sending to wrong user
"""
from MicrosoftTeams import send_proactive_message_command
import pytest
# Mock get_user to return multiple users
requests_mock.get(
f"{GRAPH_BASE_URL}/v1.0/users",
json={
"value": [
{"id": "user-1", "mail": "john@example.com", "displayName": "John Doe"},
{"id": "user-2", "mail": "john.doe@example.com", "displayName": "John Doe"},
]
},
complete_qs=False, # Allow partial query string matching
)
mocker.patch.object(demisto, "args", return_value={"user_id": "John Doe", "message": "Test"})
# Verify ValueError is raised
with pytest.raises(ValueError) as exc_info:
send_proactive_message_command()
error_msg = str(exc_info.value)
assert "Multiple users found" in error_msg
assert "To avoid sending sensitive messages to the wrong user" in error_msg
def test_send_proactive_message_command_cached_conversation(mocker, requests_mock):
"""
Given:
- A user identifier for a user with an existing cached conversation
When:
- Executing the 'microsoft-teams-send-proactive-message' command
Then:
- Verify cached conversation is reused (no new conversation created)
- Verify message is sent to cached conversation
"""
from MicrosoftTeams import send_proactive_message_command
user_id = "test-user-id-12345"
cached_conversation_id = "19:cached-conversation-id@unq.gbl.spaces"
# Setup integration context with cached conversation
cached_context = integration_context.copy()
cached_context["proactive_conversations"] = json.dumps({user_id: cached_conversation_id})
mocker.patch("MicrosoftTeams.get_integration_context", return_value=cached_context)
mocker.patch("MicrosoftTeams.set_integration_context") # Prevent actual context updates
# Mock get_user
requests_mock.get(
f"{GRAPH_BASE_URL}/v1.0/users", json={"value": [{"id": user_id, "mail": "test@example.com"}]}, complete_qs=False
)
# Mock send message (should NOT create new conversation)
send_message_mock = requests_mock.post(
f"{service_url}/v3/conversations/{cached_conversation_id}/activities", json={"id": "activity-id"}
)
mocker.patch.object(demisto, "args", return_value={"user_id": "test@example.com", "message": "Hello again!"})
return_results = mocker.patch("MicrosoftTeams.return_results")
send_proactive_message_command()
# Verify no conversation creation request was made (only send message request)
create_conversation_requests = [
req
for req in requests_mock.request_history
if "/v3/conversations" in req.url and req.method == "POST" and not req.url.endswith("/activities")
]
assert len(create_conversation_requests) == 0
# Verify message was sent to cached conversation
assert send_message_mock.called
results = return_results.call_args[0][0]
assert results.outputs["ConversationId"] == cached_conversation_id
def test_send_proactive_message_command_validation_errors(mocker, requests_mock):
"""
Given:
- Invalid argument combinations
When:
- Executing the 'microsoft-teams-send-proactive-message' command
Then:
- Verify appropriate validation errors are raised
"""
from MicrosoftTeams import send_proactive_message_command
import pytest
# Test: No message or adaptive_card provided
mocker.patch.object(demisto, "args", return_value={"user_id": "test@example.com"})
with pytest.raises(ValueError) as exc_info:
send_proactive_message_command()
assert "Either message or adaptive_card must be provided" in str(exc_info.value)
# Test: Both message and adaptive_card provided
mocker.patch.object(demisto, "args", return_value={"user_id": "test@example.com", "message": "Test", "adaptive_card": "{}"})
with pytest.raises(ValueError) as exc_info:
send_proactive_message_command()
assert "Provide either message or adaptive_card, not both" in str(exc_info.value)
def test_entitlement_handler_with_proactive_user(mocker, requests_mock):
"""
Given:
- An entitlement response from a user who received a proactive message (not in team cache)
When:
- Calling entitlement_handler
Then:
- Verify user email is retrieved from Graph API
- Verify user email is cached for future use
- Verify entitlement is handled correctly
"""
from MicrosoftTeams import entitlement_handler
proactive_user_id = "29:proactive-user-id-not-in-teams"
user_email = "proactive.user@example.com"
conversation_id = "19:proactive-conversation@unq.gbl.spaces"
activity_id = "1:activity-id"
# Mock Graph API user lookup
requests_mock.get(
f"{GRAPH_BASE_URL}/v1.0/users",
json={"value": [{"id": proactive_user_id, "mail": user_email, "userPrincipalName": user_email}]},
)
# Mock update message
requests_mock.put(f"{service_url}/v3/conversations/{conversation_id}/activities/{activity_id}", json={"id": "update-id"})
request_body = {"from": {"id": proactive_user_id}, "replyToId": activity_id}
value = {
"response": "Approved",
"entitlement": "test-entitlement-guid",
"investigation_id": "200",
"task_id": "10",
}
# Setup context without the proactive user
test_context = integration_context.copy()
test_context.pop("proactive_users", None)
mocker.patch("MicrosoftTeams.get_integration_context", return_value=test_context)
set_context_mock = mocker.patch("MicrosoftTeams.set_integration_context")
handle_entitlement_mock = mocker.patch.object(demisto, "handleEntitlementForUser")
entitlement_handler(test_context, request_body, value, conversation_id)
# Verify Graph API was called to get user email
assert any("/v1.0/users" in req.url for req in requests_mock.request_history)
# Verify user was cached
assert set_context_mock.called
cached_context = set_context_mock.call_args[0][0]
proactive_users = json.loads(cached_context.get("proactive_users", "{}"))
assert proactive_user_id in proactive_users
assert proactive_users[proactive_user_id]["email"] == user_email
# Verify entitlement was handled with correct email
assert handle_entitlement_mock.called
assert handle_entitlement_mock.call_args[1]["email"] == user_email
def test_entitlement_handler_with_cached_proactive_user(mocker, requests_mock):
"""
Given:
- An entitlement response from a user already in proactive_users cache
When:
- Calling entitlement_handler
Then:
- Verify user email is retrieved from cache (no Graph API call)
- Verify entitlement is handled correctly
"""
from MicrosoftTeams import entitlement_handler
proactive_user_id = "29:cached-proactive-user"
user_email = "cached.user@example.com"
conversation_id = "19:conversation"
activity_id = "1:activity-id"
# Setup context with cached proactive user
test_context = integration_context.copy()
test_context["proactive_users"] = json.dumps({proactive_user_id: {"email": user_email}})
# Mock update message
requests_mock.put(f"{service_url}/v3/conversations/{conversation_id}/activities/{activity_id}", json={"id": "update-id"})
request_body = {"from": {"id": proactive_user_id}, "replyToId": activity_id}
value = {
"response": "Denied",
"entitlement": "test-entitlement-guid-2",
"investigation_id": "201",
"task_id": "11",
}
mocker.patch("MicrosoftTeams.get_integration_context", return_value=test_context)
handle_entitlement_mock = mocker.patch.object(demisto, "handleEntitlementForUser")
entitlement_handler(test_context, request_body, value, conversation_id)
# Verify NO Graph API call was made (email retrieved from cache)
graph_api_calls = [req for req in requests_mock.request_history if "/v1.0/users" in req.url]
assert len(graph_api_calls) == 0
# Verify entitlement was handled with cached email
assert handle_entitlement_mock.called
assert handle_entitlement_mock.call_args[1]["email"] == user_email
def test_resolve_user_id_for_proactive_message_by_email(mocker, requests_mock):
"""
Given:
- A user email address
When:
- Calling resolve_user_id_for_proactive_message
Then:
- Verify user is resolved correctly via Graph API
"""
from MicrosoftTeams import resolve_user_id_for_proactive_message
user_email = "john.doe@example.com"
user_id = "resolved-user-id-12345"
requests_mock.get(
f"{GRAPH_BASE_URL}/v1.0/users",
json={"value": [{"id": user_id, "mail": user_email, "displayName": "John Doe"}]},
)
result = resolve_user_id_for_proactive_message(user_email)
assert result == user_id
def test_resolve_user_id_for_proactive_message_by_display_name(mocker, requests_mock):
"""
Given:
- A user display name
When:
- Calling resolve_user_id_for_proactive_message
Then:
- Verify user is resolved correctly
"""
from MicrosoftTeams import resolve_user_id_for_proactive_message
display_name = "Jane Smith"
user_id = "resolved-user-id-67890"
requests_mock.get(
f"{GRAPH_BASE_URL}/v1.0/users",
json={"value": [{"id": user_id, "mail": "jane.smith@example.com", "displayName": display_name}]},
)
result = resolve_user_id_for_proactive_message(display_name)
assert result == user_id
def test_create_proactive_conversation_new(mocker, requests_mock):
"""
Given:
- A user ID without an existing cached conversation
When:
- Calling create_proactive_conversation
Then:
- Verify new conversation is created
- Verify conversation ID is cached
"""
from MicrosoftTeams import create_proactive_conversation
user_id = "new-user-id"
new_conversation_id = "19:new-conversation"
test_context = integration_context.copy()
test_context["proactive_conversations"] = json.dumps({})
mocker.patch("MicrosoftTeams.get_integration_context", return_value=test_context)
set_context_mock = mocker.patch("MicrosoftTeams.set_integration_context")
# Mock conversation creation
requests_mock.post(f"{service_url}/v3/conversations", json={"id": new_conversation_id})
result = create_proactive_conversation(user_id)
assert result == new_conversation_id
# Verify conversation was cached
assert set_context_mock.called
updated_context = set_context_mock.call_args[0][0]
cached_conversations = json.loads(updated_context["proactive_conversations"])
assert user_id in cached_conversations
assert cached_conversations[user_id] == new_conversation_id
def test_create_proactive_conversation_cached(mocker):
"""
Given:
- A user ID with an existing cached conversation
When:
- Calling create_proactive_conversation
Then:
- Verify cached conversation ID is returned
- Verify no new conversation is created (no API call, no set_integration_context)
"""
from MicrosoftTeams import create_proactive_conversation
user_id = "cached-user-id"
cached_conversation_id = "19:cached-conversation@unq.gbl.spaces"
test_context = integration_context.copy()
test_context["proactive_conversations"] = json.dumps({user_id: cached_conversation_id})
mocker.patch("MicrosoftTeams.get_integration_context", return_value=test_context)
set_context_mock = mocker.patch("MicrosoftTeams.set_integration_context")
result = create_proactive_conversation(user_id)
assert result == cached_conversation_id
# On a cache hit, no context write should occur (no new state to persist)
set_context_mock.assert_not_called()
def test_send_proactive_message_to_conversation_with_message(mocker, requests_mock):
"""
Given:
- A conversation ID and a text message
When:
- Calling send_proactive_message_to_conversation
Then:
- Verify message is sent with correct payload
"""
from MicrosoftTeams import send_proactive_message_to_conversation
conversation_id = "19:test-conversation"
message = "Test proactive message"
activity_id = "activity-123"
send_mock = requests_mock.post(f"{service_url}/v3/conversations/{conversation_id}/activities", json={"id": activity_id})
result = send_proactive_message_to_conversation(conversation_id, message=message)
assert result["id"] == activity_id
sent_payload = send_mock.last_request.json()
assert sent_payload["type"] == "message"
assert sent_payload["text"] == message
assert sent_payload["entities"] == []
def test_send_proactive_message_to_conversation_with_adaptive_card(mocker, requests_mock):
"""
Given:
- A conversation ID and an adaptive card
When:
- Calling send_proactive_message_to_conversation
Then:
- Verify adaptive card is sent with correct payload
"""
from MicrosoftTeams import send_proactive_message_to_conversation
conversation_id = "19:test-conversation"
adaptive_card = {
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {"type": "AdaptiveCard", "version": "1.0", "body": [{"type": "TextBlock", "text": "Card message"}]},
}
send_mock = requests_mock.post(f"{service_url}/v3/conversations/{conversation_id}/activities", json={"id": "activity-456"})
result = send_proactive_message_to_conversation(conversation_id, adaptive_card=adaptive_card)
assert result["id"] == "activity-456"
sent_payload = send_mock.last_request.json()
assert sent_payload["type"] == "message"
assert sent_payload["attachments"] == [adaptive_card]
assert "text" not in sent_payload
def test_lru_cache_get_hit_and_miss():
"""
Given:
- An LRU cache (OrderedDict) with some entries
When:
- Calling lru_cache_get with an existing key (cache hit)
- Calling lru_cache_get with a non-existing key (cache miss)
Then:
- Cache hit returns the correct value and moves the key to the end (MRU position)
- Cache miss returns None and does not modify the cache
"""
from collections import OrderedDict
from MicrosoftTeams import lru_cache_get
cache: OrderedDict = OrderedDict([("key1", "val1"), ("key2", "val2"), ("key3", "val3")])
# Cache hit: key2 should be moved to end (MRU)
result = lru_cache_get(cache, "key2")
assert result == "val2"
assert list(cache.keys()) == ["key1", "key3", "key2"] # key2 moved to end
# Cache miss: returns None, cache unchanged
result = lru_cache_get(cache, "nonexistent")
assert result is None
assert list(cache.keys()) == ["key1", "key3", "key2"] # unchanged
def test_lru_cache_set_insert_and_update():
"""
Given:
- An LRU cache (OrderedDict)
When:
- Inserting a new key
- Updating an existing key
Then:
- New key is added at the end (MRU position)
- Updated key is moved to the end (MRU position)
- Cache size does not exceed max_size
"""
from collections import OrderedDict
from MicrosoftTeams import lru_cache_set
cache: OrderedDict = OrderedDict([("key1", "val1"), ("key2", "val2")])
# Insert new key
lru_cache_set(cache, "key3", "val3", max_size=10)
assert list(cache.keys()) == ["key1", "key2", "key3"]
assert cache["key3"] == "val3"
# Update existing key (should move to end)
lru_cache_set(cache, "key1", "val1_updated", max_size=10)
assert list(cache.keys()) == ["key2", "key3", "key1"]
assert cache["key1"] == "val1_updated"
def test_lru_cache_set_eviction():
"""
Given:
- An LRU cache at maximum capacity (max_size=3)
When:
- Inserting a new entry that exceeds the max size
Then:
- The least-recently-used entry (oldest/first) is evicted
- The new entry is added at the end (MRU position)
- Cache size stays at max_size
"""
from collections import OrderedDict
from MicrosoftTeams import lru_cache_set
cache: OrderedDict = OrderedDict([("key1", "val1"), ("key2", "val2"), ("key3", "val3")])
# Insert 4th entry into a max_size=3 cache → key1 (LRU) should be evicted
lru_cache_set(cache, "key4", "val4", max_size=3)
assert len(cache) == 3
assert "key1" not in cache # LRU entry evicted
assert list(cache.keys()) == ["key2", "key3", "key4"]
assert cache["key4"] == "val4"
def test_lru_cache_set_eviction_after_access():
"""
Given:
- An LRU cache at maximum capacity (max_size=3)
- key1 is accessed (making key2 the LRU)
When:
- Inserting a new entry that exceeds the max size
Then:
- key2 (now the LRU) is evicted, not key1
"""
from collections import OrderedDict
from MicrosoftTeams import lru_cache_get, lru_cache_set
cache: OrderedDict = OrderedDict([("key1", "val1"), ("key2", "val2"), ("key3", "val3")])
# Access key1 → key1 moves to end, key2 becomes LRU
lru_cache_get(cache, "key1")
assert list(cache.keys()) == ["key2", "key3", "key1"]
# Insert key4 → key2 (LRU) should be evicted
lru_cache_set(cache, "key4", "val4", max_size=3)
assert len(cache) == 3
assert "key2" not in cache # LRU entry evicted
assert "key1" in cache # recently accessed, should survive
assert list(cache.keys()) == ["key3", "key1", "key4"]
def test_lru_cache_max_size_500_enforced():
"""
Given:
- An LRU cache filled with exactly PROACTIVE_CACHE_MAX_SIZE (500) entries
When:
- Adding one more entry
Then:
- The oldest entry is evicted
- Cache size remains at 500
"""
from collections import OrderedDict
from MicrosoftTeams import lru_cache_set, PROACTIVE_CACHE_MAX_SIZE
cache: OrderedDict = OrderedDict()
for i in range(PROACTIVE_CACHE_MAX_SIZE):
cache[f"user-{i}"] = f"conv-{i}"
assert len(cache) == PROACTIVE_CACHE_MAX_SIZE
# Add one more entry using the default max_size (PROACTIVE_CACHE_MAX_SIZE)
lru_cache_set(cache, "user-new", "conv-new")
assert len(cache) == PROACTIVE_CACHE_MAX_SIZE
assert "user-0" not in cache # oldest entry evicted
assert "user-new" in cache
assert cache["user-new"] == "conv-new"
def test_proactive_conversations_lru_eviction_on_new_conversation(mocker, requests_mock):
"""
Given:
- A proactive_conversations cache already at PROACTIVE_CACHE_MAX_SIZE (500) entries
When:
- create_proactive_conversation is called for a new user
Then:
- A new conversation is created via the API
- The oldest cached entry is evicted
- The new entry is added to the cache
- Cache size stays at PROACTIVE_CACHE_MAX_SIZE
"""
from collections import OrderedDict
from MicrosoftTeams import create_proactive_conversation, PROACTIVE_CACHE_MAX_SIZE
# Build a full cache with 500 entries
full_cache: OrderedDict = OrderedDict()
for i in range(PROACTIVE_CACHE_MAX_SIZE):
full_cache[f"user-{i}"] = f"conv-{i}"
oldest_user = "user-0"
new_user_id = "brand-new-user"
new_conversation_id = "19:brand-new-conversation"
test_context = integration_context.copy()
test_context["proactive_conversations"] = json.dumps(full_cache)
mocker.patch("MicrosoftTeams.get_integration_context", return_value=test_context)
set_context_mock = mocker.patch("MicrosoftTeams.set_integration_context")
requests_mock.post(f"{service_url}/v3/conversations", json={"id": new_conversation_id})
result = create_proactive_conversation(new_user_id)
assert result == new_conversation_id
# Verify the cache was updated correctly
assert set_context_mock.called
updated_context = set_context_mock.call_args[0][0]
cached_conversations = json.loads(updated_context["proactive_conversations"])
assert len(cached_conversations) == PROACTIVE_CACHE_MAX_SIZE
assert oldest_user not in cached_conversations # LRU evicted
assert new_user_id in cached_conversations
assert cached_conversations[new_user_id] == new_conversation_id
def test_proactive_users_lru_eviction_on_new_user(mocker, requests_mock):
"""
Given:
- A proactive_users cache already at PROACTIVE_CACHE_MAX_SIZE (500) entries
When:
- entitlement_handler is called for a new user not in team cache or proactive_users cache
Then:
- User email is fetched from Graph API
- The oldest cached entry is evicted
- The new entry is added to the cache
- Cache size stays at PROACTIVE_CACHE_MAX_SIZE
"""
from collections import OrderedDict
from MicrosoftTeams import entitlement_handler, PROACTIVE_CACHE_MAX_SIZE
# Build a full proactive_users cache with 500 entries
full_cache: OrderedDict = OrderedDict()
for i in range(PROACTIVE_CACHE_MAX_SIZE):
full_cache[f"user-{i}"] = {"email": f"user{i}@example.com"}
oldest_user = "user-0"
new_proactive_user_id = "29:brand-new-proactive-user"
new_user_email = "new.proactive@example.com"
conversation_id = "19:test-conversation"
activity_id = "1:activity-id"
# Mock Graph API user lookup
requests_mock.get(
f"{GRAPH_BASE_URL}/v1.0/users",
json={"value": [{"id": new_proactive_user_id, "mail": new_user_email, "userPrincipalName": new_user_email}]},
)
# Mock update message
requests_mock.put(f"{service_url}/v3/conversations/{conversation_id}/activities/{activity_id}", json={"id": "update-id"})
test_context = integration_context.copy()
test_context["proactive_users"] = json.dumps(full_cache)
mocker.patch("MicrosoftTeams.get_integration_context", return_value=test_context)
set_context_mock = mocker.patch("MicrosoftTeams.set_integration_context")
mocker.patch.object(demisto, "handleEntitlementForUser")
request_body = {"from": {"id": new_proactive_user_id}, "replyToId": activity_id}
value = {
"response": "Approved",
"entitlement": "test-entitlement-guid",
"investigation_id": "300",
"task_id": "15",
}
entitlement_handler(test_context, request_body, value, conversation_id)
# Verify the cache was updated correctly
assert set_context_mock.called
# Find the call that updated proactive_users
proactive_users_context = None
for call in set_context_mock.call_args_list:
ctx = call[0][0]
if "proactive_users" in ctx:
proactive_users_context = ctx
break
assert proactive_users_context is not None
cached_users = json.loads(proactive_users_context["proactive_users"])
assert len(cached_users) == PROACTIVE_CACHE_MAX_SIZE
assert oldest_user not in cached_users # LRU evicted
assert new_proactive_user_id in cached_users
assert cached_users[new_proactive_user_id]["email"] == new_user_email