import MimecastV2 import pytest from freezegun import freeze_time from CommonServerPython import * QUERY_XML = """ attachmentcount status subject size receiveddate displayfrom id displayto smash displaytoaddresslist displayfromaddress """ # Parameters for Get arguments test policy_data = { "description": "new", "fromPart": "bla bla", "fromType": "free_mail_domains", "fromValue": "gmail.com", "toType": "email_domain", "toValue": "gmail.com", "option": "no_action", "policy_id": "IDFROMMIMECAST", } policy_args = { "description": "new", "fromPart": "bla bla", "fromType": "free_mail_domains", "fromValue": "gmail.com", "toType": "email_domain", "toValue": "gmail.com", "bidirectional": "", "comment": "", "enabled": "", "enforced": "", "override": "", "toDate": "", "fromDate": "", "fromEternal": "", "toEternal": "", } get_args_response = (policy_args, "no_action") # Parameters for Update policy test policy_obj = { "description": "new new", "from": {"emailDomain": "gmail.com", "type": "free_mail_domains"}, "to": {"emailDomain": "gmail.com", "type": "email_domain"}, } update_two_args = {"fromType": "free_mail_domains", "description": "new new"} update_all_args = { "fromType": "free_mail_domains", "fromValue": "gmail.com", "toType": "email_domain", "toValue": "gmail.com", "description": "new new", } update_policy_req_response = {"policy": policy_obj, "option": "no_action", "id": "IDFROMMIMECAST"} set_empty_value_args_res_list = [update_two_args, "no_action", "IDFROMMIMECAST"] set_empty_value_args_res_list_all = [update_all_args, "no_action", "IDFROMMIMECAST"] demisto_args = {"policy_id": "IDFROMMIMECAST"} MimecastV2.BASE_URL = "http://test.com" MimecastV2.APP_KEY = "test_key" MimecastV2.EMAIL_ADDRESS = "test@test.com" MimecastV2.APP_ID = "1234" MimecastV2.ACCESS_KEY = "12345" MimecastV2.SECRET_KEY = "test_key==" def util_load_json(path): """ Args: path: path to load json from. Returns: json object read from the path given """ with open(path, encoding="utf-8") as f: return json.loads(f.read()) def test_get_arguments_for_policy_command(): res = MimecastV2.get_arguments_for_policy_command(policy_data) assert get_args_response == res def test_update_policy(mocker): mocker.patch.object(MimecastV2, "get_arguments_for_policy_command", return_value=get_args_response) mocker.patch.object(MimecastV2, "set_empty_value_args_policy_update", return_value=set_empty_value_args_res_list) mocker.patch.object(MimecastV2, "create_or_update_policy_request", return_value=update_policy_req_response) mocker.patch.object(demisto, "args", return_value=demisto_args) res = MimecastV2.update_policy_command() assert res["Contents"]["Description"] == "new new" assert res["Contents"]["Sender"]["Type"] == "free_mail_domains" mocker.patch.object(MimecastV2, "get_arguments_for_policy_command", return_value=get_args_response) mocker.patch.object(MimecastV2, "set_empty_value_args_policy_update", return_value=set_empty_value_args_res_list_all) mocker.patch.object(MimecastV2, "create_or_update_policy_request", return_value=update_policy_req_response) mocker.patch.object(demisto, "args", return_value=demisto_args) res = MimecastV2.update_policy_command() assert res["Contents"]["Description"] == "new new" assert res["Contents"]["Sender"]["Type"] == "free_mail_domains" assert res["Contents"]["Sender"]["Domain"] == "gmail.com" assert res["Contents"]["Receiver"]["Type"] == "email_domain" assert res["Contents"]["Receiver"]["Domain"] == "gmail.com" INCIDENT_API_RESPONSE = { "fail": [], "meta": {"status": 200}, "data": [ { "code": "TR-CSND1A7780-00045-M", "successful": 0, "create": "2020-05-25T10:01:53+0000", "modified": "2020-05-25T10:01:53+0000", "identified": 3, "failed": 0, "reason": "test", "id": "test-id", "type": "manual", "searchCriteria": { "start": "2020-04-25T10:01:53+0000", "end": "2020-05-25T22:01:53+0000", "messageId": "test message id", }, "restored": 0, } ], } EXPECTED_MARKDOWN_RESPONSE = """### Incident test-id has been created #### Code: TR-CSND1A7780-00045-M #### Type: manual #### Reason: test #### The number of messages identified based on the search criteria: 3 #### The number successfully remediated messages: 0 #### The number of messages that failed to remediate: 0 #### The number of messages that were restored from the incident: 0 |End date|Message ID| |---|---| | 2020-05-25T22:01:53+0000 | test message id | """ def test_mimecast_incident_api_response_to_markdown(): actual_response = MimecastV2.mimecast_incident_api_response_to_markdown(INCIDENT_API_RESPONSE, "create") assert actual_response == EXPECTED_MARKDOWN_RESPONSE EXPECTED_CONTEXT_RESPONSE = { "Mimecast.Incident(val.ID && val.ID == obj.ID)": { "Reason": "test", "Code": "TR-CSND1A7780-00045-M", "FailedRemediatedMessages": 0, "IdentifiedMessages": 3, "MessagesRestored": 0, "LastModified": "2020-05-25T10:01:53+0000", "SearchCriteria": { "StartDate": "2020-04-25T10:01:53+0000", "EndDate": "2020-05-25T22:01:53+0000", "FileHash": None, "To": None, "MessageID": "test message id", "From": None, }, "Type": "manual", "ID": "test-id", "SuccessfullyRemediatedMessages": 0, } } def test_mimecast_incident_api_response_to_context(): actual_response = MimecastV2.mimecast_incident_api_response_to_context(INCIDENT_API_RESPONSE) assert actual_response == EXPECTED_CONTEXT_RESPONSE add_member_req_response = {"data": [{"emailAddress": "test@gmail.com", "folderId": "folder_id"}]} get_group_members_req_response = {"data": [{"groupMembers": {}}]} def test_mimecast_add_remove_member_to_group_with_email(mocker): """Unit test Given - add_remove_member_to_group command - command args - email and group id. - command raw response When - mock the server response to create_add_remove_group_member_request. - mock the server response to create_get_group_members_request Then Validate the content of the HumanReadable. """ mocker.patch.object(demisto, "args", return_value={"group_id": "1234", "email_address": "test@gmail.com"}) mocker.patch.object(MimecastV2, "create_add_remove_group_member_request", return_value=add_member_req_response) mocker.patch.object(MimecastV2, "create_get_group_members_request", return_value=get_group_members_req_response) result = MimecastV2.add_remove_member_to_group("add") assert result.readable_output == "test@gmail.com had been added to group ID folder_id" add_member_req_response_no_email = {"data": [{"folderId": "folder_id"}]} def test_mimecast_add_remove_member_to_group_with_domain(mocker): """Unit test Given - add_remove_member_to_group command - command args - domain and group id. - command raw response When - mock the server response to create_add_remove_group_member_request. - mock the server response to create_get_group_members_request Then Validate the content of the HumanReadable. """ mocker.patch.object(demisto, "args", return_value={"group_id": "1234", "domain": "test.com"}) mocker.patch.object(MimecastV2, "create_add_remove_group_member_request", return_value=add_member_req_response_no_email) mocker.patch.object(MimecastV2, "create_get_group_members_request", return_value=get_group_members_req_response) results = MimecastV2.add_remove_member_to_group("add") assert results.readable_output == "Address had been added to group ID folder_id" CREATE_MANAGED_URL_SUCCESSFUL_MOCK = { "fail": [], "meta": {"status": 200}, "data": [ { "comment": "None", "domain": "www.test.com", "queryString": "", "disableRewrite": False, "port": -1, "disableUserAwareness": False, "disableLogClick": False, "action": "permit", "path": "", "matchType": "explicit", "scheme": "https", "id": "fake_id", } ], } def test_create_managed_url(mocker): """Unit test Given - create_managed_url command - the url does not exist - command args - url, action, matchType, disableRewrite, disableUserAwareness, disableLogClick - command raw response When - mock the server response to create_managed_url_request. Then Validate the content of the command result. """ args = { "url": "https://www.test.com", "action": "permit", "matchType": "explicit", "disableRewrite": "false", "disableUserAwareness": "false", "disableLogClick": "false", } expected_context = { "Mimecast.URL(val.ID && val.ID == obj.ID)": [ { "Domain": "www.test.com", "disableRewrite": False, "disableLogClick": False, "Action": "permit", "Path": "", "matchType": "explicit", "ID": "fake_id", } ] } mocker.patch.object(demisto, "args", return_value=args) mocker.patch.object(MimecastV2, "create_managed_url_request", return_value=CREATE_MANAGED_URL_SUCCESSFUL_MOCK["data"][0]) results = MimecastV2.create_managed_url() assert "Managed URL https://www.test.com created successfully!" in results.get("HumanReadable") assert expected_context == results.get("EntryContext") def test_add_users_under_group_in_context_dict__dict(mocker): """ Given - Users list - Group id _ Integration context with `group` key with a single group in it When - adding users under group in context dict as part of `mimecast-get-group-members` command Then Returns a valid outputs """ context = {"Mimecast": {"Group": {"ID": "groupID", "Users": []}}} users_list = [ { "Domain": "demistodev.com", "Name": "", "EmailAddress": "testing@demistodev.com", "InternalUser": True, "Type": "created_manually", "IsRemoved": False, } ] expected = [ { "ID": "groupID", "Users": [ { "Domain": "demistodev.com", "Name": "", "EmailAddress": "testing@demistodev.com", "InternalUser": True, "Type": "created_manually", "IsRemoved": False, } ], } ] mocker.patch.object(demisto, "context", return_value=context) result = MimecastV2.add_users_under_group_in_context_dict(users_list, "groupID") assert result == expected def test_add_users_under_group_in_context_dict__list(mocker): """ Given - Users list - Group id _ Integration context with `group` key with list of groups in it When - adding users under group in context dict as part of `mimecast-get-group-members` command Then Returns a valid outpus """ context = {"Mimecast": {"Group": [{"ID": "groupID", "Users": []}, {"ID": "groupID2", "Users": []}]}} users_list = [ { "Domain": "demistodev.com", "Name": "", "EmailAddress": "testing@demistodev.com", "InternalUser": True, "Type": "created_manually", "IsRemoved": False, } ] expected = [ { "ID": "groupID", "Users": [ { "Domain": "demistodev.com", "Name": "", "EmailAddress": "testing@demistodev.com", "InternalUser": True, "Type": "created_manually", "IsRemoved": False, } ], }, {"ID": "groupID2", "Users": []}, ] mocker.patch.object(demisto, "context", return_value=context) result = MimecastV2.add_users_under_group_in_context_dict(users_list, "groupID") assert result == expected def test_search_message_command(mocker): """ Given: - Message id to search. When: - Running a search message command to retrieve information on given message. Then: - Make sure search data is returned. """ mock_response = util_load_json("test_data/search_message_response.json") mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) args = {"message_id": "12345"} response = MimecastV2.search_message_command(args) assert response.outputs == mock_response.get("data")[0].get("trackedEmails") assert response.outputs_prefix == "Mimecast.SearchMessage" assert response.outputs_key_field == "id" def test_held_message_summary_command(mocker): """ When: - Running a hold message summary command to retrieve hold information messages. Then: - Make sure hold data is returned. """ mock_response = util_load_json("test_data/hold_message_summary_response.json") mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) response = MimecastV2.held_message_summary_command() assert response.outputs == mock_response.get("data") assert response.outputs_prefix == "Mimecast.HeldMessageSummary" assert response.outputs_key_field == "policyInfo" MESSAGE_INFO_ARGS = [ ({"ids": "12345, 1345", "show_delivered_message": "true"}, True, 1), ({"ids": "12345, 1345", "show_delivered_message": "false"}, False, 0), ] @pytest.mark.parametrize("args, delivered, delivered_message_len", MESSAGE_INFO_ARGS) def test_get_message_info_command(args, delivered, delivered_message_len, requests_mock): """ Given: - Message ids to get info for. When: - Running a get message info to retrieve information for. Then: - Make sure correct data is returned. """ mock_response = util_load_json("test_data/get_message_info_response.json") requests_mock.post("/api/message-finder/get-message-info", json=mock_response) response = MimecastV2.get_message_info_command(args) assert len(response) == 2 assert ("test@test.com" in response[0].readable_output) == delivered assert isinstance(response[0].outputs.get("deliveredMessage"), list) assert len(response[0].outputs.get("deliveredMessage")) == delivered_message_len assert response[0].outputs_prefix == "Mimecast.MessageInfo" def test_list_held_messages_command(mocker): """ When: - Running a list hold messages command. Then: - Make sure correct data is returned. """ mock_response = util_load_json("test_data/list_hold_messages_response.json") mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) args = {"admin": "true", "limit": "10"} response = MimecastV2.list_held_messages_command(args) assert len(response.outputs) == 10 assert response.outputs == mock_response.get("data") assert response.outputs_prefix == "Mimecast.HeldMessage" assert response.outputs_key_field == "id" @pytest.mark.parametrize( "field_name,expected_field_name", [ ("all", "all"), ("subject", "subject"), ("sender", "sender"), ("recipient", "recipient"), ("reasonCode", "reasonCode"), ("senderIP", "senderIP"), ("reason_code", "reasonCode"), # Backward compatibility: old value should be converted ], ) def test_list_held_messages_all_field_names(mocker, field_name, expected_field_name): """ Test that all valid field_name options are accepted and processed correctly, including backward compatibility. Given: - Arguments with different field_name values (all, subject, sender, recipient, reasonCode, senderIP, reason_code). When: - Running list_held_messages_request function. Then: - Ensure each field_name is accepted and passed correctly to the API. - Ensure backward compatibility: 'reason_code' is automatically converted to 'reasonCode'. """ args = { "admin": "true", "from_date": "2023-01-01T00:00:00+0000", "to_date": "2023-12-31T23:59:59+0000", "value": "test_value", "field_name": field_name, "limit": "10", } mock_response = ([{"id": "123", "subject": "test"}], 1) mock_request = mocker.patch.object(MimecastV2, "request_with_pagination", return_value=mock_response) # Call the function result = MimecastV2.list_held_messages_request(args) # Verify the field_name is used correctly call_args = mock_request.call_args data_param = call_args[1]["data"][0] assert data_param["searchBy"]["fieldName"] == expected_field_name assert result == mock_response REJECT_HOLD_MESSAGE = [ ( {"meta": {"status": 200}, "data": [{"id": "1234", "reject": True}, {"id": "1233", "reject": True}], "fail": []}, "Held message with id 1234 was rejected successfully.\nHeld message with id 1233 was rejected successfully.\n", False, ), ({"meta": {"status": 200}, "data": [{"id": "1234", "reject": False}, {"id": "1233", "reject": True}], "fail": []}, "", True), ] @pytest.mark.parametrize("mock_response, readable_output, is_exception_raised", REJECT_HOLD_MESSAGE) def test_reject_held_message_command(mock_response, readable_output, is_exception_raised, mocker): """ When: - Running a reject hold messages command. Then: - Make sure correct data is returned. """ mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) args = {"ids": "1234,1233", "message": "test", "reason_type": "MESSAGE CONTAINS UNDESIRABLE CONTENT", "notify": "true"} try: response = MimecastV2.reject_held_message_command(args) assert response.readable_output == readable_output except Exception: assert is_exception_raised RELEASE_HOLD_MESSAGE = [ ( {"meta": {"status": 200}, "data": [{"id": "1234", "release": True}], "fail": []}, "Held message with id 1234 was released successfully", False, ), ({"meta": {"status": 200}, "data": [{"id": "1234", "release": False}], "fail": []}, "Message release has failed.", True), ] @pytest.mark.parametrize("mock_response, readable_output, is_exception_raised", RELEASE_HOLD_MESSAGE) def test_release_held_message_command(mock_response, readable_output, is_exception_raised, mocker): """ When: - Running a release hold messages command. Then: - Make sure correct data is returned. """ mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) args = {"id": "1234"} try: response = MimecastV2.release_held_message_command(args) assert response.readable_output == readable_output except Exception: assert is_exception_raised def test_search_processing_message_command(mocker): """ When: - Running a search processing message command to retrieve information regarding messages being proccessed. Then: - Make sure hold data is returned. """ mock_response = util_load_json("test_data/search_processing_message_response.json") mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) args = {"sort_order": "ascending", "from_date": "2015-11-16T14:49:18+0000", "to_date": "2021-11-16T14:49:18+0000"} response = MimecastV2.search_processing_message_command(args) assert response.outputs == mock_response.get("data")[0].get("messages") assert response.outputs_prefix == "Mimecast.ProcessingMessage" assert response.outputs_key_field == "id" def test_list_email_queues_command(mocker): """ When: - Running a search processing message command to retrieve information regarding messages being proccessed. Then: - Make sure hold data is returned. """ mock_response = util_load_json("test_data/search_processing_message_response.json") mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) args = {"sort_order": "ascending", "from_date": "2015-11-16T14:49:18+0000", "to_date": "2021-11-16T14:49:18+0000"} response = MimecastV2.search_processing_message_command(args) assert response.outputs == mock_response.get("data")[0].get("messages") assert response.outputs_prefix == "Mimecast.ProcessingMessage" assert response.outputs_key_field == "id" def test_parse_queried_fields(): assert MimecastV2.parse_queried_fields(QUERY_XML) == ( "attachmentcount", "status", "subject", "size", "receiveddate", "displayfrom", "id", "displayto", "smash", "displaytoaddresslist", "displayfromaddress", ) def test_query(mocker): """ Test case for the 'query' function of the MimecastV2 integration. GIVEN: - a mocked HTTP request to Mimecast API with query data, WHEN: - 'query' function is called with the provided arguments, THEN: - Make sure all return-field values are returned to context and human-readable.""" query_data = util_load_json("test_data/query_response.json") mocker.patch.object(MimecastV2, "http_request", return_value=query_data["response"]) result = MimecastV2.query({"queryXml": QUERY_XML}) assert result["HumanReadable"] == ( "### Mimecast archived emails\n" "|Subject|Display From|Display To|Received Date|Size|Attachment Count|Status|ID|displayfromaddress|displaytoaddresslist|" "smash|\n" "|---|---|---|---|---|---|---|---|---|---|---|\n" "| Netting | test1 | test1 | 2023-08-06T07:23:00+0000 | 2262 | 0 | ARCHIVED | test1_id | test1 | {'displayableName': ''," " 'emailAddress': 'test1'} | test1_smash |\n" "| RE | test2 | test2 | 2023-08-06T07:23:00+0000 | 11370 | 0 | ARCHIVED | test2_id | test2 | {'displayableName': ''," " 'emailAddress': 'test2'} | test2_smash |\n" "| Re | test3 | test3 | 2023-08-06T07:23:00+0000 | 5280 | 0 | ARCHIVED | test3_id | test3 | {'displayableName': ''," " 'emailAddress': 'test3'} | test3_smash |\n" ) assert result["Contents"] == query_data["query_contents"] def test_empty_query(mocker): """ Test case for the 'query' function of the MimecastV2 integration, where no args are given. GIVEN: - a mocked HTTP request to Mimecast API with query data, WHEN: - 'query' function is called with no queryXml argument, THEN: - Make sure no exception is raised. """ error = { "field": "query", "code": "err_validation_blank", "message": "This field, if present, cannot be blank or empty", "retryable": False, } def query_mocked_api( api_endpoint: str, data: list, response_param: str = None, limit: int = 100, page: int = None, page_size: int = None, use_headers: bool = False, is_file: bool = False, ): if not data[0].get("query"): raise Exception(json.dumps(error)) else: return [], None mocker.patch.object(MimecastV2, "request_with_pagination", side_effect=query_mocked_api) results = MimecastV2.query({}) assert len(results.get("Contents")) == 0 def test_query_email_arguments(requests_mock): """ Test case for the 'query' function of the MimecastV2 integration with email arguments. GIVEN: - A mocked HTTP request to Mimecast API with specific email parameters. WHEN: - The 'query' function is called with sentTo, sentFrom, and subject arguments. THEN: - Ensure the request body matches the expected XML structure with the correct values. """ query_data = util_load_json("test_data/query_response.json") expected_body = { "admin": True, "query": ' \n' ' \n' ' \n' " \n" " \n" " attachmentcount\n" " status\n" " subject\n" " size\n" " receiveddate\n" " displayfrom\n" " id\n" " displayto\n" " smash\n" " \n" " \n" " \n" " subject: Test Email Subject\n" ' \n' ' sender@example.comrecipient@example.com\n' ' \n' " \n" " \n" "", } mocked_post = requests_mock.post(f"{MimecastV2.BASE_URL}/api/archive/search", json=query_data["response"], status_code=200) args = {"sentTo": "recipient@example.com", "sentFrom": "sender@example.com", "subject": "Test Email Subject"} MimecastV2.query(args) sent_body = mocked_post.last_request.json() assert sent_body["data"] == [expected_body], "The request body does not match the expected structure." def test_get_message_metadata_with_attachments(mocker): """ Given: Message metadata from API with attachments. When: Get message command being called. Then: Verify the extension value is returned. """ message_id = "123" expected_metadata = { "subject": "Test Email", "from": {"emailAddress": "test@example.com"}, "attachments": [{"filename": "hello world", "id": "1", "extension": ".txt"}], } mocker.patch.object(MimecastV2, "http_request", return_value={"data": [expected_metadata]}) _, actual_metadata = MimecastV2.get_message_metadata(message_id) assert expected_metadata.get("subject") == actual_metadata.get("Subject") assert expected_metadata.get("from").get("emailAddress") == actual_metadata.get("From") assert expected_metadata.get("attachments")[0].get("filename") == actual_metadata.get("Attachments")[0].get("FileName") assert expected_metadata.get("attachments")[0].get("extension") == actual_metadata.get("Attachments")[0].get("Extension") assert expected_metadata.get("attachments")[0].get("id") == actual_metadata.get("Attachments")[0].get("ID") def test_get_archive_search_logs_command(mocker): """ Test case for the 'get_archive_search_logs_command' function of the MimecastV2 class, where no 'query_xml' argument is given. GIVEN: - A mocked HTTP request to the Mimecast API (using http_request). WHEN: - The 'get_archive_search_logs_command' function is called without the 'query_xml' argument. THEN: - Make sure no exception is raised. """ args = {"limit": "5", "query": "integration.com"} mock_response = util_load_json("test_data/get_archive_search_logs_response.json") mocker.patch.object(MimecastV2, "request_with_pagination", return_value=mock_response) result = MimecastV2.get_archive_search_logs_command(args) assert result.outputs["data"][0]["logs"] == mock_response[0]["data"][0]["logs"] assert result.outputs_prefix == "Mimecast.ArchiveSearchLog" def test_get_search_logs_command(mocker): """ Tests the 'get_archive_search_logs_command' function of the MimecastV2 class with various arguments. This test mocks the http_request method to return a sample response containing archive search logs data. It then calls the get_archive_search_logs_command function with arguments specifying limit, page, page_size, query, and start date. Finally, it asserts that the extracted logs data matches the expected response. Args: mocker (pytest.MonkeyFixture): Pytest mocker fixture used to patch methods. """ args = {"limit": "50", "page": "1", "page_size": "1", "query": "aa.aa.aa.aa", "start": "2017-09-16T14:49:18+0000"} mock_response = util_load_json("test_data/get_search_logs_response.json") mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) result = MimecastV2.get_archive_search_logs_command(args) assert result.outputs == mock_response.get("data")[0]["logs"] assert result.outputs_prefix == "Mimecast.ArchiveSearchLog" def test_get_view_logs_command(mocker): """ Test the get_view_logs_command function of the MimecastV2 integration. Args: mocker (pytest_mock.plugin.MockerFixture): Pytest mocker fixture. """ args = {"end": "2024-09-16T14:49:18+0000", "limit": "2", "query": "example@test.com", "start": "2017-09-16T14:49:18+0000"} mock_response = util_load_json("test_data/get_view_logs_response.json") mocker.patch.object(MimecastV2, "request_with_pagination", return_value=mock_response) result = MimecastV2.get_view_logs_command(args) assert result.outputs["data"] == mock_response[0].get("data") assert result.outputs_prefix == "Mimecast.ViewLog" def test_list_account_command(mocker): """ Test the list_account_command function of the MimecastV2 integration. Args: mocker (pytest_mock.plugin.MockerFixture): Pytest mocker fixture. """ args = {"account_code": "ABC123"} mock_response = util_load_json("test_data/list_account_response.json") mocker.patch.object(MimecastV2, "request_with_pagination", return_value=mock_response) result = MimecastV2.list_account_command(args) assert result.outputs["data"] == mock_response[0].get("data") assert result.outputs_prefix == "Mimecast.Account" assert result.outputs_key_field == "accountCode" @pytest.mark.parametrize( "args, mock_response, expected_outputs_prefix", [ ( {"limit": "1", "page": "1", "page_size": "1", "policyType": "antispoofing-bypass"}, (util_load_json("test_data/list_policies_response.json"), 1), "Mimecast.AntispoofingBypassPolicy", ), ( {"limit": "1", "page": "1", "page_size": "1", "policyType": "address-alteration"}, (util_load_json("test_data/list_policies_response.json"), 1), "Mimecast.AddressAlterationPolicy", ), ], ) def test_list_policies_command(mocker, args, mock_response, expected_outputs_prefix): """ Unit test for the list_policies_command function in MimecastV2 integration (v1 policy types). Given - list_policies_command function from MimecastV2 integration. - command args including policyType set to a non-blockedsenders (v1) type. - command raw response from the server. When - mock the server response to request_with_pagination. Then - Validate the content of the response to match the expected data structure. Args: mocker (pytest_mock.plugin.MockerFixture): Pytest mocker fixture. """ mocker.patch.object(MimecastV2, "request_with_pagination", return_value=mock_response) result = MimecastV2.list_policies_command(args) assert result.outputs == mock_response[0] assert result.outputs_prefix == expected_outputs_prefix assert result.outputs_key_field == "id" def test_create_antispoofing_bypass_policy_command(mocker): """ When: Running create antispoofing bypass policy command. Args: mocker (pytest_mock.plugin.MockerFixture): Pytest mocker fixture. """ args = { "bidirectional": "no", "comment": "test", "description": "test", "enabled": "no", "enforced": "no", "from_date": "1 day", "from_eternal": "no", "from_part": "envelope_from", "from_type": "email_domain", "from_value": "googl.com", "option": "disable_bypass", "override": "no", "spf_domain": "google.com", "to_date": "now", "to_eternal": "no", "to_type": "email_domain", "to_value": "google.com", } mock_response = util_load_json("test_data/create_antispoofing_bypass_policy_response.json") mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) result = MimecastV2.create_antispoofing_bypass_policy_command(args) id = mock_response["data"][0]["id"] assert result.outputs == mock_response.get("data") assert result.outputs_prefix == "Mimecast.AntispoofingBypassPolicy" assert result.readable_output == f"Anti-Spoofing Bypass policy {id} was created successfully" assert result.outputs_key_field == "id" def test_update_antispoofing_bypass_policy_command(mocker): """ Given - update_antispoofing_bypass_policy_command function from MimecastV2 integration. - command args including description, enabled, from_eternal, id, option, and to_eternal. - command raw response from the server. When - mock the server response to http_request. Then - Validate the content of the response to match the expected data structure. - Ensure the readable output contains the correct policy ID and success message. - Check the outputs_prefix is correct. Args: mocker (pytest_mock.plugin.MockerFixture): Pytest mocker fixture. """ args = { "bidirectional": "no", "description": "test", "enabled": "no", "from_date": "1 day", "from_eternal": "no", "from_part": "both", "option": "disable_bypass", "policy_id": "eNo1jr0Ogj", "to_date": "now", "to_eternal": "yes", } mock_response = util_load_json("test_data/update_antispoofing_bypass_policy_response.json") mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) result = MimecastV2.update_antispoofing_bypass_policy_command(args) assert mock_response.get("data") == result.outputs assert f'Policy ID- {args["policy_id"]} has been updated successfully.' == result.readable_output assert result.outputs_prefix == "Mimecast.AntispoofingBypassPolicy" assert result.outputs_key_field == "id" def test_update_address_alteration_policy_command(mocker): """Unit test Given - update_address_alteration_policy_command function from MimecastV2 integration. - command args including enabled, enforced, from_eternal, from_type, policy_description, policy_id, to_eternal, and to_type. - command raw response from the server. When - mock the server response to http_request. Then - Validate the content of the response to match the expected data structure. """ args = { "bidirectional": "no", "comment": "test-comment", "conditions": "8.8.8.8/24", "enabled": "no", "enforced": "no", "from_date": "1 day", "from_eternal": "no", "from_part": "envelope_from", "override": "no", "policy_description": "test", "policy_id": "eNo1jrsOgjA", "to_date": "now", "to_eternal": "no", } id = args["policy_id"] mock_response = util_load_json("test_data/update_address_alteration_policy_response.json") mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) result = MimecastV2.update_address_alteration_policy_command(args) assert result.outputs == mock_response.get("data") assert result.readable_output == f"{id} has been updated successfully" assert result.outputs_prefix == "Mimecast.AddressAlterationPolicy" assert result.outputs_key_field == "id" def test_get_message_body_content_request_with_mailbox(mocker): """ Test case for get_message_body_content_request with mailbox parameter. Given: - message_id, message_context set to 'DELIVERED', message_type, and mailbox parameter. When: - get_message_body_content_request is called. Then: - Ensure the mailbox is included in the request payload. - Ensure the API request is made with the correct data. """ message_id = "test_message_id" message_context = "DELIVERED" message_type = "html" mailbox = "user@example.com" mock_response = type("obj", (object,), {"content": b"test content"})() http_request_mock = mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) result = MimecastV2.get_message_body_content_request(message_id, message_context, message_type, mailbox=mailbox) # Verify the API was called with the correct data call_args = http_request_mock.call_args assert call_args[0][0] == "POST" assert call_args[0][1] == "/api/archive/get-message-part" payload = call_args[0][2] assert payload["data"][0]["id"] == message_id assert payload["data"][0]["context"] == message_context assert payload["data"][0]["type"] == message_type assert payload["data"][0]["mailbox"] == mailbox assert result == b"test content" def test_get_message_body_content_request_without_mailbox_delivered(): """ Test case for get_message_body_content_request without mailbox when context is DELIVERED. Given: - message_id, message_type, and message_context set to 'DELIVERED', but no mailbox parameter. When: - get_message_body_content_request is called. Then: - Ensure a ValueError is raised with the appropriate error message. """ message_id = "test_message_id" message_context = "DELIVERED" message_type = "html" with pytest.raises(ValueError, match="The 'mailbox' parameter is required when context is set to 'DELIVERED'"): MimecastV2.get_message_body_content_request(message_id, message_context, message_type) def test_get_message_body_content_request_without_mailbox_received(mocker): """ Test case for get_message_body_content_request without mailbox when context is RECEIVED. Given: - message_id, message_type, and message_context set to 'RECEIVED', no mailbox parameter. When: - get_message_body_content_request is called. Then: - Ensure the request is made successfully without requiring mailbox. - Ensure mailbox is not included in the request payload. """ message_id = "test_message_id" message_context = "RECEIVED" message_type = "html" mock_response = type("obj", (object,), {"content": b"test content"})() http_request_mock = mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) result = MimecastV2.get_message_body_content_request(message_id, message_context, message_type) # Verify the API was called with the correct data call_args = http_request_mock.call_args assert call_args[0][0] == "POST" assert call_args[0][1] == "/api/archive/get-message-part" payload = call_args[0][2] assert payload["data"][0]["id"] == message_id assert payload["data"][0]["context"] == message_context assert payload["data"][0]["type"] == message_type assert "mailbox" not in payload["data"][0] assert result == b"test content" def test_get_message_command_with_mailbox(mocker): """ Test case for get_message command with mailbox parameter. Given: - messageID, type, context set to 'DELIVERED', and mailbox in args. When: - get_message command is called. Then: - Ensure the mailbox parameter is passed to get_message_body_content_request. """ args = { "messageID": "test_message_id", "context": "DELIVERED", "type": "html", "part": "message", "mailbox": "user@example.com", } mock_content = b"test content" get_message_body_mock = mocker.patch.object(MimecastV2, "get_message_body_content_request", return_value=mock_content) mocker.patch.object(demisto, "args", return_value=args) mocker.patch("MimecastV2.fileResult", return_value={"Type": 1, "File": "test"}) MimecastV2.get_message() # Verify get_message_body_content_request was called with mailbox get_message_body_mock.assert_called_once_with("test_message_id", "DELIVERED", "html", "user@example.com") class TestGenerateLogId: """Tests for the generate_log_id function.""" def test_generate_log_id_with_native_id(self): """Test that native ID is used when available.""" log_entry = {"id": "native123", "fileName": "test.zip"} result = MimecastV2.generate_log_id(log_entry, "attachment") assert result == "native123" def test_generate_log_id_url_without_native_id(self): """Test URL log ID generation without native ID.""" log_entry = {"url": "http://malicious.com", "date": "2024-01-15T10:00:00+0000", "userEmailAddress": "user@test.com"} result = MimecastV2.generate_log_id(log_entry, "url") assert result == "http://malicious.com_2024-01-15T10:00:00+0000_user@test.com" def test_generate_log_id_attachment_without_native_id(self): """Test attachment log ID generation without native ID.""" log_entry = { "fileName": "malware.zip", "date": "2024-01-15T10:00:00+0000", "senderAddress": "sender@test.com", "recipientAddress": "recipient@test.com", } result = MimecastV2.generate_log_id(log_entry, "attachment") assert result == "malware.zip_2024-01-15T10:00:00+0000_sender@test.com_recipient@test.com" def test_generate_log_id_impersonation_without_native_id(self): """Test impersonation log ID generation without native ID (uses date field).""" log_entry = {"subject": "Urgent Request", "date": "2024-01-15T10:00:00+0000", "senderAddress": "fake@test.com"} result = MimecastV2.generate_log_id(log_entry, "impersonation") assert result == "Urgent Request_2024-01-15T10:00:00+0000_fake@test.com" def test_generate_log_id_fallback_to_hash(self): """Test that hash is used as fallback for held_message and unknown types.""" log_entry = { "subject": "Spam Message", "dateReceived": "2024-01-15T10:00:00+0000", "from": {"emailAddress": "spam@test.com"}, } result = MimecastV2.generate_log_id(log_entry, "held_message") # Same entry should generate same hash result2 = MimecastV2.generate_log_id(log_entry, "held_message") assert result == result2 class TestFetchLogsWithPagination: """Tests for the generic fetch_logs_with_pagination function.""" def test_fetch_logs_with_pagination_no_dedup(self, mocker): """Test fetching logs without deduplication.""" mock_response = { "fail": [], "meta": {"pagination": {"next": ""}}, "data": [ { "clickLogs": [ {"id": "log1", "url": "http://test1.com", "date": "2024-01-01T10:00:00+0000"}, {"id": "log2", "url": "http://test2.com", "date": "2024-01-01T11:00:00+0000"}, ] } ], } mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) results, count, next_page = MimecastV2.fetch_logs_with_pagination( api_endpoint="/api/ttp/url/get-logs", data=[{"from": "2024-01-01T00:00:00+0000"}], response_param="clickLogs", limit=10, dedup_messages=[], current_next_page="", ) assert len(results) == 2 assert count == 2 assert next_page == "" assert results[0]["id"] == "log1" assert results[1]["id"] == "log2" def test_fetch_logs_with_pagination_with_dedup(self, mocker): """Test fetching logs with deduplication.""" mock_response = { "fail": [], "meta": {"pagination": {"next": ""}}, "data": [ { "clickLogs": [ {"id": "log1", "url": "http://test1.com"}, {"id": "log2", "url": "http://test2.com"}, {"id": "log3", "url": "http://test3.com"}, ] } ], } mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) results, count, next_page = MimecastV2.fetch_logs_with_pagination( api_endpoint="/api/ttp/url/get-logs", data=[{"from": "2024-01-01T00:00:00+0000"}], response_param="clickLogs", limit=10, dedup_messages=["log1", "log3"], # log1 and log3 are duplicates current_next_page="", ) assert len(results) == 1 assert count == 1 assert results[0]["id"] == "log2" def test_fetch_logs_with_pagination_direct_data_access(self, mocker): """Test fetching logs with direct data access (no response_param).""" mock_response = { "fail": [], "meta": {"pagination": {"next": ""}}, "data": [ {"id": "msg1", "subject": "Test 1", "dateReceived": "2024-01-01T10:00:00+0000"}, {"id": "msg2", "subject": "Test 2", "dateReceived": "2024-01-01T11:00:00+0000"}, ], } mocker.patch.object(MimecastV2, "http_request", return_value=mock_response) results, count, next_page = MimecastV2.fetch_logs_with_pagination( api_endpoint="/api/gateway/get-hold-message-list", data=[{"start": "2024-01-01T00:00:00+0000", "admin": True}], response_param=None, limit=10, dedup_messages=[], current_next_page="", ) assert len(results) == 2 assert count == 2 assert results[0]["id"] == "msg1" assert results[1]["id"] == "msg2" class TestFetchLogType: """Tests for the generic fetch_log_type function.""" @freeze_time("2024-01-15T12:00:00Z") def test_fetch_log_type_url_first_run(self, mocker): """Test fetching URL logs for the first time.""" mock_logs = [ {"id": "url1", "url": "http://malicious.com", "date": "2024-01-15T10:00:00+0000"}, {"id": "url2", "url": "http://bad.com", "date": "2024-01-15T11:00:00+0000"}, ] mocker.patch.object(MimecastV2, "fetch_logs_with_pagination", return_value=(mock_logs, 2, "")) last_run = {} current_fetch = datetime(2024, 1, 15, 10, 0, 0) incidents = [] new_last_run = {"time": "2024-01-15T12:00:00Z"} MimecastV2.fetch_log_type( log_type="url", api_endpoint="/api/ttp/url/get-logs", response_param="clickLogs", search_params={"from": "2024-01-15T10:00:00+0000", "scanResult": "malicious"}, to_incident_func=MimecastV2.url_to_incident, last_run=last_run, current_fetch=current_fetch, incidents=incidents, new_last_run=new_last_run, ) assert len(incidents) == 2 assert "time_url" in new_last_run assert "dedup_url" in new_last_run @freeze_time("2024-01-15T12:00:00Z") def test_fetch_log_type_with_pagination_continuation(self, mocker): """Test fetching logs with pagination continuation.""" mock_logs = [ {"id": "url5", "url": "http://test.com", "date": "2024-01-15T11:30:00+0000"}, ] mocker.patch.object(MimecastV2, "fetch_logs_with_pagination", return_value=(mock_logs, 1, "next_token_456")) last_run = { "time_url": "2024-01-15T11:00:00Z", "url_next_page": "next_token_123", "time_url_for_next_page": "2024-01-15T11:00:00Z", "dedup_url": [], } current_fetch = datetime(2024, 1, 15, 11, 0, 0) incidents = [] new_last_run = {"time": "2024-01-15T12:00:00Z"} MimecastV2.fetch_log_type( log_type="url", api_endpoint="/api/ttp/url/get-logs", response_param="clickLogs", search_params={"from": "2024-01-15T11:00:00+0000", "scanResult": "malicious"}, to_incident_func=MimecastV2.url_to_incident, last_run=last_run, current_fetch=current_fetch, incidents=incidents, new_last_run=new_last_run, ) assert len(incidents) == 1 assert "url_next_page" in new_last_run assert new_last_run["url_next_page"] == "next_token_456" class TestFetchIncidents: """Tests for the main fetch_incidents function.""" @freeze_time("2024-01-15T12:00:00Z") def test_fetch_incidents_first_run(self, mocker): """Test fetch_incidents on first run.""" mocker.patch.object(MimecastV2, "FETCH_URL", True) mocker.patch.object(MimecastV2, "FETCH_ATTACHMENTS", False) mocker.patch.object(MimecastV2, "FETCH_IMPERSONATIONS", False) mocker.patch.object(MimecastV2, "FETCH_HELD_MESSAGES", False) mocker.patch.object(MimecastV2, "MAX_FETCH", 100) mocker.patch.object(demisto, "getLastRun", return_value={}) set_last_run_mock = mocker.patch.object(demisto, "setLastRun") incidents_mock = mocker.patch.object(demisto, "incidents") mocker.patch.object(MimecastV2, "fetch_log_type") MimecastV2.fetch_incidents() assert set_last_run_mock.called assert incidents_mock.called @freeze_time("2024-01-15T12:00:00Z") def test_fetch_incidents_with_previous_state(self, mocker): """Test fetch_incidents with previous state.""" previous_last_run = { "time": "2024-01-15T10:00:00Z", "time_url": "2024-01-15T10:00:00Z", "dedup_url": ["url1"], "url_next_page": "token123", "time_url_for_next_page": "2024-01-15T10:00:00Z", } mocker.patch.object(MimecastV2, "FETCH_URL", True) mocker.patch.object(MimecastV2, "FETCH_ATTACHMENTS", False) mocker.patch.object(MimecastV2, "FETCH_IMPERSONATIONS", False) mocker.patch.object(MimecastV2, "FETCH_HELD_MESSAGES", False) mocker.patch.object(demisto, "getLastRun", return_value=previous_last_run) set_last_run_mock = mocker.patch.object(demisto, "setLastRun") mocker.patch.object(demisto, "incidents") mocker.patch.object(MimecastV2, "fetch_log_type") MimecastV2.fetch_incidents() assert set_last_run_mock.called class TestHttpRequestErrorHandling: """Tests for the v2 error-envelope parsing in http_request.""" def test_success_returns_json(self, requests_mock): """200 response returns parsed JSON.""" requests_mock.get("http://test.com/api/test", json={"key": "value"}) result = MimecastV2.http_request("GET", "/api/test") assert result == {"key": "value"} def test_is_file_returns_response_object(self, requests_mock): """is_file=True returns the raw response object, not json().""" requests_mock.get("http://test.com/api/test", content=b"binary") result = MimecastV2.http_request("GET", "/api/test", is_file=True) assert result.content == b"binary" def test_v2_error_envelope_raises_demisto_exception(self, requests_mock): """400 with v2 error envelope raises DemistoException with the message.""" requests_mock.patch( "http://test.com/api/test", status_code=400, json={"error": [{"code": "err_test", "message": "Something went wrong"}]}, ) with pytest.raises(DemistoException, match="Something went wrong"): MimecastV2.http_request("PATCH", "/api/test", payload={}) def test_v2_multiple_errors_joined(self, requests_mock): """Multiple errors in the v2 envelope are joined with '; '.""" requests_mock.patch( "http://test.com/api/test", status_code=400, json={ "error": [ {"code": "err_a", "message": "First error"}, {"code": "err_b", "message": "Second error"}, ] }, ) with pytest.raises(DemistoException, match="First error; Second error"): MimecastV2.http_request("PATCH", "/api/test", payload={}) def test_non_v2_4xx_reraises_http_error(self, requests_mock): """4xx without v2 error key re-raises the original HTTPError.""" import requests requests_mock.patch( "http://test.com/api/test", status_code=403, json={"message": "Forbidden"}, ) with pytest.raises(requests.exceptions.HTTPError): MimecastV2.http_request("PATCH", "/api/test", payload={}) def test_json_parse_failure_reraises_http_error(self, requests_mock): """If the error body is not JSON, re-raises the original HTTPError.""" import requests requests_mock.patch( "http://test.com/api/test", status_code=400, text="not json", headers={"Content-Type": "text/plain"}, ) with pytest.raises(requests.exceptions.HTTPError): MimecastV2.http_request("PATCH", "/api/test", payload={}) class TestUpdateBlockSenderPolicyCommand: """Tests for update_block_sender_policy_command.""" def test_missing_policy_id_raises(self): with pytest.raises(DemistoException, match="policy ID"): MimecastV2.update_block_sender_policy_command({}) def test_success_returns_readable_output(self, requests_mock): requests_mock.patch( "http://test.com/policy-management/cloud-gateway/v1/blocked-senders/policies/abc123", status_code=204, text="" ) result = MimecastV2.update_block_sender_policy_command({"policy_id": "abc123", "description": "updated"}) assert "abc123" in result.readable_output assert result.outputs is None def test_patch_body_contains_from_object(self, requests_mock): adapter = requests_mock.patch( "http://test.com/policy-management/cloud-gateway/v1/blocked-senders/policies/pid", status_code=204, text="", ) MimecastV2.update_block_sender_policy_command( {"policy_id": "pid", "fromType": "email_domain", "fromValue": "example.com"} ) sent_body = adapter.last_request.json() assert sent_body["from"] == {"type": "email_domain", "domain": "example.com"} class TestOAuth2TokenManagement: """Tests for token_oauth2_request and updating_token_oauth2.""" # --- token_oauth2_request --- def test_token_oauth2_request_returns_token_and_expires_in(self, requests_mock): """Successful token fetch returns (access_token, expires_in) tuple.""" requests_mock.post( "http://test.com/oauth/token", json={"access_token": "tok123", "expires_in": 1799, "token_type": "Bearer", "scope": ""}, ) token, expires_in = MimecastV2.token_oauth2_request() assert token == "tok123" assert expires_in == 1799 def test_token_oauth2_request_defaults_expires_in_when_missing(self, requests_mock): """If expires_in is absent from the response, defaults to 1799.""" requests_mock.post( "http://test.com/oauth/token", json={"access_token": "tok456", "token_type": "Bearer"}, ) token, expires_in = MimecastV2.token_oauth2_request() assert token == "tok456" assert expires_in == 1799 # --- updating_token_oauth2 --- def test_updating_token_fetches_when_no_context(self, mocker, requests_mock): """Fetches a new token when integration context is empty.""" requests_mock.post( "http://test.com/oauth/token", json={"access_token": "new_tok", "expires_in": 1799, "token_type": "Bearer", "scope": ""}, ) mocker.patch.object(demisto, "getIntegrationContext", return_value={}) set_ctx = mocker.patch.object(demisto, "setIntegrationContext") MimecastV2.updating_token_oauth2() assert MimecastV2.TOKEN_OAUTH2 == "new_tok" ctx = set_ctx.call_args[0][0] assert ctx["value"] == "new_tok" assert ctx["expires_in"] == 1799 assert "last_update" in ctx @freeze_time("2024-01-15T12:00:00Z") def test_updating_token_reuses_valid_token(self, mocker): """Does not fetch a new token when the existing one is still valid.""" now = MimecastV2.epoch_seconds() mocker.patch.object( demisto, "getIntegrationContext", return_value={"value": "cached_tok", "last_update": now - 60, "expires_in": 1799}, ) fetch_mock = mocker.patch.object(MimecastV2, "token_oauth2_request") MimecastV2.updating_token_oauth2() fetch_mock.assert_not_called() assert MimecastV2.TOKEN_OAUTH2 == "cached_tok" @freeze_time("2024-01-15T12:00:00Z") def test_updating_token_refreshes_when_expired(self, mocker, requests_mock): """Fetches a new token when the existing one has expired.""" requests_mock.post( "http://test.com/oauth/token", json={"access_token": "refreshed_tok", "expires_in": 1799, "token_type": "Bearer", "scope": ""}, ) now = MimecastV2.epoch_seconds() mocker.patch.object( demisto, "getIntegrationContext", return_value={"value": "old_tok", "last_update": now - 1799, "expires_in": 1799}, ) mocker.patch.object(demisto, "setIntegrationContext") MimecastV2.updating_token_oauth2() assert MimecastV2.TOKEN_OAUTH2 == "refreshed_tok" @freeze_time("2024-01-15T12:00:00Z") def test_updating_token_uses_expires_in_from_context(self, mocker): """Uses the expires_in stored in context, not a hardcoded value.""" now = MimecastV2.epoch_seconds() mocker.patch.object( demisto, "getIntegrationContext", return_value={"value": "short_ttl_tok", "last_update": now - 200, "expires_in": 300}, ) fetch_mock = mocker.patch.object(MimecastV2, "token_oauth2_request") MimecastV2.updating_token_oauth2() fetch_mock.assert_not_called() assert MimecastV2.TOKEN_OAUTH2 == "short_ttl_tok" @freeze_time("2024-01-15T12:00:00Z") def test_updating_token_refreshes_within_safety_margin(self, mocker, requests_mock): """Refreshes when fewer than 60s remain before expiry.""" requests_mock.post( "http://test.com/oauth/token", json={"access_token": "margin_tok", "expires_in": 1799, "token_type": "Bearer", "scope": ""}, ) now = MimecastV2.epoch_seconds() mocker.patch.object( demisto, "getIntegrationContext", return_value={"value": "expiring_tok", "last_update": now - 1750, "expires_in": 1799}, ) mocker.patch.object(demisto, "setIntegrationContext") MimecastV2.updating_token_oauth2() assert MimecastV2.TOKEN_OAUTH2 == "margin_tok" BLOCKED_SENDERS_V2_URL = "http://test.com/policy-management/cloud-gateway/v1/blocked-senders/policies" V2_FLAT_POLICY = { "id": "9f0d1a2b-3c4d-5e6f-7a8b-9c0d1e2f3a4b", "description": "Block example.com", "from": {"type": "email_domain", "domain": "example.com"}, "to": {"type": "everyone"}, "bidirectional": False, "fromDateTime": "2024-01-15T12:00:00+00:00", "toDateTime": "2024-02-15T12:00:00+00:00", "fromEternal": False, "toEternal": True, "fromPart": "envelope_from", "enabled": True, "enforced": False, "override": False, } class TestBuildPolicyV2Bodies: """Tests for the v2 request-body builders shared by create and update.""" def test_nested_from_to_by_type(self): """fromValue/toValue land in domain/emailAddress/groupId according to the type.""" body = MimecastV2.build_blocked_senders_policy_v2_body( { "fromType": "email_domain", "fromValue": "example.com", "toType": "individual_email_address", "toValue": "user@test.com", } ) assert body["from"] == {"type": "email_domain", "domain": "example.com"} assert body["to"] == {"type": "individual_email_address", "emailAddress": "user@test.com"} def test_group_id_mapping(self): """profile_group maps the value to groupId.""" body = MimecastV2.build_blocked_senders_policy_v2_body({"fromType": "profile_group", "fromValue": "grp-1"}) assert body["from"] == {"type": "profile_group", "groupId": "grp-1"} def test_valueless_type_emits_type_only(self): """A type that takes no value emits only the type key.""" body = MimecastV2.build_blocked_senders_policy_v2_body({"toType": "everyone"}) assert body["to"] == {"type": "everyone"} @freeze_time("2024-01-15T12:00:00Z") def test_date_args_renamed_to_date_time(self): """from_date/to_date are renamed to fromDateTime/toDateTime, keeping the v1 date format.""" from_date, to_date = "1 day", "now" expected_from = arg_to_datetime(from_date).strftime(MimecastV2.DATE_FORMAT) # type: ignore[union-attr] expected_to = arg_to_datetime(to_date).strftime(MimecastV2.DATE_FORMAT) # type: ignore[union-attr] body = MimecastV2.build_blocked_senders_policy_v2_body({"from_date": from_date, "to_date": to_date}) assert body["fromDateTime"] == expected_from assert body["toDateTime"] == expected_to assert "fromDate" not in body assert "toDate" not in body def test_patch_body_omits_unprovided_fields(self): """Partial-update semantics: nothing the user did not provide is sent.""" body = MimecastV2.build_blocked_senders_policy_v2_body({"description": "only this"}) assert body == {"description": "only this"} class TestCreateBlockSenderPolicyCommand: """Tests for create_block_sender_policy_command (v2 POST).""" def test_success_returns_id_in_readable_output(self, requests_mock): """201 {id} produces the success HR and no context.""" requests_mock.post(BLOCKED_SENDERS_V2_URL, status_code=201, json={"id": "new-uuid"}) result = MimecastV2.create_block_sender_policy_command( {"description": "d", "option": "block_sender", "fromType": "email_domain", "fromValue": "example.com"} ) assert result.readable_output == "Policy new-uuid was created successfully." assert result.outputs == {"id": "new-uuid"} def test_request_body_is_v2_nested(self, requests_mock): """The POST body uses the nested v2 schema with top-level scalars.""" adapter = requests_mock.post(BLOCKED_SENDERS_V2_URL, status_code=201, json={"id": "new-uuid"}) MimecastV2.create_block_sender_policy_command( { "description": "Block example.com", "option": "block_sender", "fromPart": "envelope_from", "fromType": "email_domain", "fromValue": "example.com", "toType": "everyone", } ) sent_body = adapter.last_request.json() assert sent_body == { "description": "Block example.com", "option": "block_sender", "fromPart": "envelope_from", "from": {"type": "email_domain", "domain": "example.com"}, "to": {"type": "everyone"}, } def test_v2_error_envelope_surfaces_message(self, requests_mock): """A v2 error envelope is surfaced as a DemistoException.""" requests_mock.post( BLOCKED_SENDERS_V2_URL, status_code=400, json={"error": [{"code": "err_policy_invalid", "message": "Invalid policy"}]}, ) with pytest.raises(DemistoException, match="Invalid policy"): MimecastV2.create_block_sender_policy_command({"description": "d"}) class TestGetPolicyCommandV2: """Tests for the blockedsenders branch of get_policy_command (v2 per-id GET).""" def test_emits_flat_object_under_single_prefix(self, requests_mock): """The bare flat object is emitted verbatim under Mimecast.BlockedSendersPolicy.""" policy_id = V2_FLAT_POLICY["id"] requests_mock.get(f"{BLOCKED_SENDERS_V2_URL}/{policy_id}", json=V2_FLAT_POLICY) result = MimecastV2.get_policy_command({"policyID": policy_id, "policyType": "blockedsenders"}) assert result.outputs == V2_FLAT_POLICY assert result.outputs_prefix == "Mimecast.BlockedSendersPolicy" def test_hr_uses_corrected_receiver_spelling(self, requests_mock): """The HR table uses 'Receiver', not the legacy 'Reciever'.""" policy_id = V2_FLAT_POLICY["id"] requests_mock.get(f"{BLOCKED_SENDERS_V2_URL}/{policy_id}", json=V2_FLAT_POLICY) result = MimecastV2.get_policy_command({"policyID": policy_id, "policyType": "blockedsenders"}) assert "Receiver" in result.readable_output assert "Reciever" not in result.readable_output assert "example.com" in result.readable_output def test_missing_policy_id_raises(self): with pytest.raises(DemistoException, match="policy ID"): MimecastV2.get_policy_command({"policyType": "blockedsenders"}) def test_v1_policy_type_keeps_legacy_behavior(self, mocker): """A non-blockedsenders type still uses the v1 request and dual-prefix outputs.""" mocker.patch.object( MimecastV2, "get_policy_request", return_value=[{"id": "v1-id", "policy": {"from": {}, "to": {}, "fromDate": "d1", "toDate": "d2"}}], ) results = MimecastV2.get_policy_command({"policyID": "v1-id", "policyType": "antispoofing-bypass"}) assert [r.outputs_prefix for r in results] == ["Mimecast.Policy", "Mimecast.AntispoofingBypassPolicy"] class TestListPoliciesCommandV2: """Tests for the blockedsenders branch of list_policies_command (v2 GET list).""" def test_emits_policies_array_verbatim(self, requests_mock): """response['policies'] is emitted as-is, with no reshaping.""" requests_mock.get(BLOCKED_SENDERS_V2_URL, json={"policies": [V2_FLAT_POLICY], "meta": {"nextPage": "cursor"}}) result = MimecastV2.list_policies_command({"policyType": "blockedsenders"}) assert result.outputs == {"policies": [V2_FLAT_POLICY], "NextToken": "cursor"} assert result.outputs_prefix == "Mimecast.BlockedSendersPolicy" def test_empty_policies_list(self, requests_mock): """An empty result set does not raise.""" requests_mock.get(BLOCKED_SENDERS_V2_URL, json={"policies": [], "meta": {}}) result = MimecastV2.list_policies_command({"policyType": "blockedsenders"}) assert result.outputs == {} def test_hr_uses_corrected_receiver_spelling(self, requests_mock): """The HR table uses 'Receiver', not the legacy 'Reciever'.""" requests_mock.get(BLOCKED_SENDERS_V2_URL, json={"policies": [V2_FLAT_POLICY], "meta": {}}) result = MimecastV2.list_policies_command({"policyType": "blockedsenders"}) assert "Receiver" in result.readable_output assert "Reciever" not in result.readable_output def test_defaults_to_blockedsenders(self, requests_mock): """With no policyType the command uses the v2 blockedsenders endpoint.""" requests_mock.get(BLOCKED_SENDERS_V2_URL, json={"policies": [V2_FLAT_POLICY], "meta": {}}) result = MimecastV2.list_policies_command({}) assert result.outputs == {"policies": [V2_FLAT_POLICY]} class TestDeletePolicyCommandV2: """Tests for the blockedsenders branch of delete_policy (v2 DELETE).""" def test_204_is_treated_as_success(self, requests_mock): """An empty 204 body is a success; context is built from the input ID.""" requests_mock.delete(f"{BLOCKED_SENDERS_V2_URL}/pid", status_code=204, text="") results = MimecastV2.delete_policy({"policyID": "pid", "policyType": "blockedsenders"}) assert [r.outputs_prefix for r in results] == ["Mimecast.Policy", "Mimecast.BlockedSendersPolicy"] for result in results: assert result.outputs == {"ID": "pid", "Deleted": True} assert result.readable_output == "Mimecast Policy pid deleted successfully!" def test_uses_delete_verb_on_v2_path(self, requests_mock): """The request is a DELETE against the v2 per-id path.""" adapter = requests_mock.delete(f"{BLOCKED_SENDERS_V2_URL}/pid", status_code=204, text="") MimecastV2.delete_policy({"policyID": "pid", "policyType": "blockedsenders"}) assert adapter.last_request.method == "DELETE" assert adapter.last_request.path.endswith("/blocked-senders/policies/pid") def test_not_found_raises(self, requests_mock): """A 404 error envelope is surfaced as a DemistoException.""" requests_mock.delete( f"{BLOCKED_SENDERS_V2_URL}/missing", status_code=404, json={"error": [{"code": "err_policy_not_found", "message": "Policy not found"}]}, ) with pytest.raises(DemistoException, match="Policy not found"): MimecastV2.delete_policy({"policyID": "missing", "policyType": "blockedsenders"}) def test_v1_policy_type_keeps_legacy_request(self, mocker): """A non-blockedsenders type still routes through the v1 delete request.""" v1_request = mocker.patch.object(MimecastV2, "delete_policy_request") MimecastV2.delete_policy({"policyID": "v1-id", "policyType": "address-alteration"}) v1_request.assert_called_once_with("address-alteration", "v1-id")