import pytest from CommonServerPython import * from freezegun import freeze_time import json MOCK_BASEURL = "https://example.com" MOCK_API = "api_key" from AdminByRequestEventCollector import ( deny_request_command, approve_request_command, list_requests_command, Client, EventType, remove_first_run_params, validate_fetch_events_params, set_event_type_fetch_limit, fetch_events_list, fetch_events, get_events, EVENT_TYPES, prepare_list_output, ) @pytest.fixture def client(): """ A dummy client fixture for testing. """ return Client( base_url=MOCK_BASEURL, api_key=MOCK_API, verify=False, use_proxy=False, ) def util_load_json(path): with open(path, encoding="utf-8") as f: return json.loads(f.read()) class TestHelperFunction: audit_log_event = EVENT_TYPES["Auditlog"] events_event = EVENT_TYPES["Events"] requests_event = EVENT_TYPES["Requests"] @pytest.mark.parametrize( "input_params, expected_params", [ ({"startdate": "2024-01-01", "enddate": "2024-01-31", "other": 1}, {"other": 1}), ({"startdate": "2024-01-01", "other": 1}, {"other": 1}), ({"enddate": "2024-01-31", "other": 1}, {"other": 1}), ({"other": 1}, {"other": 1}), # Case 5: empty dict ({}, {}), ], ) def test_remove_first_run_params(self, input_params: Dict[str, Any], expected_params: Dict[str, Any]): """ Test remove_first_run_params function behavior on initial fetch. Given: - Dictionary Containing params from "last_run" When: - Checking if to update the params or not Then: - Make sure the correct params are coming back - not containing "first run" params" """ # make a copy to avoid modifying original test data params_copy = input_params.copy() remove_first_run_params(params_copy) assert params_copy == expected_params def test_prepare_list_output(self): """ Given: A list of records. When: Calling the prepare_list_output function. Then: Ensure the output is a markdown table as expected. """ records = [ {"id": "1", "status": "open", "user": "test"}, {"id": "2", "status": "closed", "user": "test"}, ] output = prepare_list_output(records) assert "|ID|Status|User|" in output assert "|---|---|---|" in output assert "| 1 | open | test |" in output assert "| 2 | closed | test |" in output output = prepare_list_output([]) assert "No entries." in output result_param = {"startid": 1, "take": 1} case1_validate_fetch_events_params = ( ({"start_id_auditlog": 1}, audit_log_event, False), ({**audit_log_event.default_params, **result_param}, "auditlog", "start_id_auditlog"), ) case2_validate_fetch_events_params = ( ({"start_id_events": 1}, events_event, False), ({**events_event.default_params, **result_param}, "events", "start_id_events"), ) case3_validate_fetch_events_params = ( ({"start_id_requests": 1}, requests_event, False), ({**requests_event.default_params, **result_param}, "requests", "start_id_requests"), ) # cases where we dont have key then use date 2025-01-01 01:00:00 date_params = {"startdate": "2025-01-01", "enddate": "2025-01-01"} result_param_no_start_id = {"take": 1} case4_validate_fetch_events_params = ( ({}, audit_log_event, False), ({**date_params, **result_param_no_start_id}, "auditlog", "start_id_auditlog"), ) case5_validate_fetch_events_params = ( ({}, events_event, False), ({**date_params, **result_param_no_start_id}, "events", "start_id_events"), ) case6_validate_fetch_events_params = ( ({}, requests_event, False), ({**requests_event.default_params, **result_param_no_start_id}, "requests", "start_id_requests"), ) # cases where we dont have key then use date 2025-01-01 01:00:00 last_run_start_date = {"startdate": "2025-01-01"} # using case7_validate_fetch_events_params = ( (last_run_start_date, audit_log_event, True), ({**last_run_start_date, **result_param_no_start_id}, "auditlog", "start_id_auditlog"), ) @pytest.mark.parametrize( "input_params, expected_results", [ case1_validate_fetch_events_params, case2_validate_fetch_events_params, case3_validate_fetch_events_params, case4_validate_fetch_events_params, case5_validate_fetch_events_params, case6_validate_fetch_events_params, case7_validate_fetch_events_params, ], ) @freeze_time("2025-01-01 01:00:00") def test_validate_fetch_events_params( self, input_params: tuple[dict, EventType, bool], expected_results: tuple[dict, str, str] ) -> None: """ Test validate_fetch_events_params function behavior on initial fetch. Given: - An input params containing params from "last_run", event type and a flag mention should we use lat run as params. When: - Validating params before fetch action. Then: - Make sure the request is sent with right parameters. """ results = validate_fetch_events_params(*input_params) assert results == expected_results case1_set_event_type_fetch_limit = ( { "event_types_to_fetch": ["Auditlog", "Events", "Requests"], "max_auditlog_per_fetch": 50000, "max_events_per_fetch": 50000, "max_requests_per_fetch": 5000, }, 3, (50000, 50000, 5000), ) case2_set_event_type_fetch_limit = ( { "event_types_to_fetch": ["Auditlog", "Events"], "max_auditlog_per_fetch": 50000, "max_events_per_fetch": 50000, "max_requests_per_fetch": 5000, }, 2, (50000, 50000), ) case3_set_event_type_fetch_limit = ( { "event_types_to_fetch": ["Auditlog", "Requests"], "max_auditlog_per_fetch": 50000, "max_events_per_fetch": 10, "max_requests_per_fetch": 10, }, 2, (50000, 10), ) case4_set_event_type_fetch_limit = ( { "event_types_to_fetch": [], "max_auditlog_per_fetch": 50000, "max_events_per_fetch": 50000, "max_requests_per_fetch": 5000, }, 0, (), ) @pytest.mark.parametrize( "input_params, expected_len, expected_limits", [ case1_set_event_type_fetch_limit, case2_set_event_type_fetch_limit, case3_set_event_type_fetch_limit, case4_set_event_type_fetch_limit, ], ) def test_set_event_type_fetch_limit( self, input_params: Dict[str, Any], expected_len: int, expected_limits: tuple[int, int, int] ) -> None: """ Test set_event_type_fetch_limit function behavior. Given: - An input params containing integration params from integration configuration. When: - VUpdating Event types before making a fetch action Then: - Make dure each event type has the correct max_fetch limit. """ event_types = set_event_type_fetch_limit(input_params) assert len(event_types) == expected_len for i in range(expected_len): assert event_types[i].max_fetch == expected_limits[i] class TestFetchEvents: event_requests = EVENT_TYPES["Requests"] event_events = EVENT_TYPES["Events"] event_audit = EVENT_TYPES["Auditlog"] raw_detections_audit = util_load_json("test_data/auditlogs_response.json") raw_detections_events = util_load_json("test_data/events_response.json") raw_detections_requests = util_load_json("test_data/requests_response.json") def test_fetch_events_update_last_run(self, client, mocker): """ Given: A mock raw response containing audit logs. When: fetching events. Then: Make sure that the last run object was updated as expected """ self.event_audit.max_fetch = 3 raw_detections = self.raw_detections_audit mocker.patch("AdminByRequestEventCollector.Client.get_events_request", return_value=raw_detections) last_run = {} output = fetch_events_list(client, last_run=last_run, event_type=self.event_audit, use_last_run_as_params=False) assert len(output) == 3 assert last_run.get("start_id_auditlog") == raw_detections[-1]["id"] + 1 def test_fetch_events_update_last_run_with_old_params(self, client, mocker): """ Given: A mock raw response containing audit logs. When: fetching events. Then: Make sure that the last run object was updated as expected """ self.event_audit.max_fetch = 3 raw_detections = self.raw_detections_audit mocker.patch("AdminByRequestEventCollector.Client.get_events_request", return_value=raw_detections) last_run = {"start_id_auditlog": 1, "start_id_events": 1, "start_id_requests": 1} output = fetch_events_list(client, last_run=last_run, event_type=self.event_audit, use_last_run_as_params=False) assert len(output) == 3 assert last_run.get("start_id_auditlog") == raw_detections[-1]["id"] + 1 assert last_run.get("start_id_events") == 1 assert last_run.get("start_id_requests") == 1 def test_fetch_response_bigger_then_limit(self, client, mocker): """ Given: A mock raw response containing events. When: Fetching events with a fetch limit smaller than the number of available logs. Then: Ensure the function returns exactly the requested number of events and updates the last run timestamp correctly. """ self.event_events.max_fetch = 2 raw_detections = self.raw_detections_events mocker.patch("AdminByRequestEventCollector.Client.get_events_request", return_value=raw_detections) last_run = {"start_id_auditlog": 1, "start_id_events": 1, "start_id_requests": 1} output = fetch_events_list(client, last_run=last_run, event_type=self.event_events, use_last_run_as_params=False) assert len(output) == 2 assert last_run.get("start_id_events") == raw_detections[-1]["id"] def test_fetch_limit_bigger_then_response(self, client, mocker): """ Given: A mock raw response containing events. When: Fetching events with a fetch limit higher than the number of available logs. Then: Ensure the function returns exactly the requested number of events and updates the last run timestamp correctly. """ self.event_requests.max_fetch = 3 raw_detections = self.raw_detections_requests[:-1] first_response = raw_detections second_response = [] mocker.patch("AdminByRequestEventCollector.Client.get_events_request", side_effect=[first_response, second_response]) last_run = {"start_id_auditlog": 1, "start_id_events": 1, "start_id_requests": 1} output = fetch_events_list(client, last_run=last_run, event_type=self.event_requests, use_last_run_as_params=False) assert len(output) == len(raw_detections) assert last_run.get("start_id_requests") == raw_detections[-1]["id"] + 1 def test_fetch_all_types(self, client, mocker): """ Given: A mock raw response containing audit logs. When: fetching events from all types of EventType Then: Ensure the function returns exactly the requested number of events and updates the last run correctly. """ self.event_requests.max_fetch = 3 self.event_events.max_fetch = 3 self.event_audit.max_fetch = 3 raw_detections = self.raw_detections_audit + self.raw_detections_events + self.raw_detections_requests first_response = self.raw_detections_audit second_response = self.raw_detections_events third_response = self.raw_detections_requests events_types = [self.event_audit, self.event_events, self.event_requests] mocker.patch( "AdminByRequestEventCollector.Client.get_events_request", side_effect=[first_response, second_response, third_response], ) output, last_run = fetch_events(client, last_run={}, fetch_events_types=events_types, use_last_run_as_params=False) assert len(output) == len(raw_detections) assert last_run.get("start_id_auditlog") == self.raw_detections_audit[-1]["id"] + 1 assert last_run.get("start_id_events") == self.raw_detections_events[-1]["id"] + 1 assert last_run.get("start_id_requests") == self.raw_detections_requests[-1]["id"] + 1 def test_fetch_all_types_different_lengths(self, client, mocker): """ Given: A mock raw response containing audit logs. When: fetching events from all types of EventType Then: Ensure the function returns exactly the requested number of events and updates the last run correctly. """ self.event_audit.max_fetch = 3 self.event_events.max_fetch = 2 self.event_requests.max_fetch = 1 raw_detections = self.raw_detections_audit + self.raw_detections_events[:-1] + self.raw_detections_requests[:-2] first_response = self.raw_detections_audit second_response = self.raw_detections_events third_response = self.raw_detections_requests events_types = [self.event_audit, self.event_events, self.event_requests] mocker.patch( "AdminByRequestEventCollector.Client.get_events_request", side_effect=[first_response, second_response, third_response], ) output, last_run = fetch_events(client, last_run={}, fetch_events_types=events_types, use_last_run_as_params=False) assert len(output) == len(raw_detections) assert last_run.get("start_id_auditlog") == self.raw_detections_audit[-1]["id"] + 1 assert last_run.get("start_id_events") == self.raw_detections_events[-2]["id"] + 1 assert last_run.get("start_id_requests") == self.raw_detections_requests[-3]["id"] + 1 def test_fetch_all_types_field_values_audits(self, client, mocker): """ Given: A mock raw response containing audit logs. When: fetching events. Then: Make sure that the special XSIAM fields was updated as expected """ self.event_audit.max_fetch = 3 raw_detections = self.raw_detections_audit mocker.patch("AdminByRequestEventCollector.Client.get_events_request", return_value=raw_detections) last_run = {} output = fetch_events_list(client, last_run=last_run, event_type=self.event_audit, use_last_run_as_params=False) for i in range(len(output)): assert output[i].get(self.event_audit.time_field) == raw_detections[i]["startTimeUTC"] assert output[i].get("source_log_type") == self.event_audit.source_log_type def test_fetch_all_types_field_values_events(self, client, mocker): """ Given: A mock raw response containing audit logs. When: fetching events. Then: Make sure that the special XSIAM fields was updated as expected """ self.event_events.max_fetch = 3 raw_detections = self.raw_detections_events mocker.patch("AdminByRequestEventCollector.Client.get_events_request", return_value=raw_detections) last_run = {} output = fetch_events_list(client, last_run=last_run, event_type=self.event_events, use_last_run_as_params=False) for i in range(len(output)): assert output[i].get(self.event_events.time_field) == raw_detections[i]["eventTimeUTC"] assert output[i].get("source_log_type") == self.event_events.source_log_type def test_fetch_all_types_field_values_requests(self, client, mocker): """ Given: A mock raw response containing audit logs. When: fetching events. Then: Make sure that the special XSIAM fields was updated as expected """ self.event_requests.max_fetch = 3 raw_detections = self.raw_detections_requests mocker.patch("AdminByRequestEventCollector.Client.get_events_request", return_value=raw_detections) last_run = {} output = fetch_events_list(client, last_run=last_run, event_type=self.event_requests, use_last_run_as_params=False) for i in range(len(output)): assert output[i].get(self.event_requests.time_field) == raw_detections[i]["requestTime"] assert output[i].get("source_log_type") == self.event_requests.source_log_type @freeze_time("2025-01-01 01:00:00") def test_list_requests_command(self, client, mocker): """ Given: A mock raw response containing requests. When: fetching requests using the list_requests_command function. Then: Make sure that output is as expected. """ raw_detections = self.raw_detections_requests args = {"limit": 2} mocker.patch("AdminByRequestEventCollector.Client.get_events_request", return_value=raw_detections) output = list_requests_command(client, args=args) assert len(output.outputs) == len(raw_detections) assert output.outputs_prefix == "AdminByRequest.Request" def test_approve_request_command(self, client, mocker): """ Given: A request ID and an approver email. When: Calling the approve_request_command function. Then: Ensure the client's approve_request method is called with the correct arguments and a success message is returned. """ args = {"request_id": "12345", "approved_by": "test@example.com"} mock_response = mocker.Mock() mock_response.status_code = 204 mocker.patch("AdminByRequestEventCollector.Client.approve_request", return_value=mock_response) result = approve_request_command(client, args) client.approve_request.assert_called_once_with("requests/12345", {"approvedby": "test@example.com"}) assert result.readable_output == "Request with 12345 id was successfully approved." def test_deny_request_command(self, client, mocker): """ Given: A request ID, a reason, and a denier email. When: Calling the deny_request_command function. Then: Ensure the client's deny_request method is called with the correct arguments and a success message is returned. """ args = {"request_id": "12345", "reason": "Security risk", "denied_by": "test@example.com"} mock_response = mocker.Mock() mock_response.status_code = 204 mocker.patch("AdminByRequestEventCollector.Client.deny_request", return_value=mock_response) result = deny_request_command(client, args) client.deny_request.assert_called_once_with("requests/12345", {"reason": "Security risk", "deniedby": "test@example.com"}) assert result.readable_output == "Request with 12345 id was successfully denied." @freeze_time("2025-01-01 01:00:00") def test_get_events(self, client, mocker): """ Given: A mock raw response containing audit logs. When: fetching events using the get events function. Then: Make sure that output is as expected. """ raw_detections = self.raw_detections_events[:-1] first_response = self.raw_detections_events second_response = [] args = {"limit": 2, "event_type": "Events"} mocker.patch("AdminByRequestEventCollector.Client.get_events_request", side_effect=[first_response, second_response]) output = get_events(client, args=args) assert len(output.outputs) == len(raw_detections) assert output.outputs_prefix == "AdminByRequest." + "Events" class TestCommandFunctions: def test_approve_request_command_fail(self, client, mocker): """ Given: A request ID and an approver email. When: Calling the approve_request_command function and the API fails. Then: Ensure a DemistoException is raised. """ args = {"request_id": "12345", "approved_by": "test@example.com"} mock_response = mocker.Mock() mock_response.status_code = 400 mocker.patch("AdminByRequestEventCollector.Client.approve_request", return_value=mock_response) with pytest.raises(DemistoException) as e: approve_request_command(client, args) assert "Failed to approve request 12345. Status code: 400" in str(e.value) def test_approve_request_command_no_id(self, client): """ Given: No request ID. When: Calling the approve_request_command function. Then: Ensure a ValueError is raised. """ with pytest.raises(ValueError) as e: approve_request_command(client, {}) assert "request_id is required" in str(e.value) def test_deny_request_command_fail(self, client, mocker): """ Given: A request ID, a reason, and a denier email. When: Calling the deny_request_command function and the API fails. Then: Ensure a DemistoException is raised. """ args = {"request_id": "12345", "reason": "Security risk", "denied_by": "test@example.com"} mock_response = mocker.Mock() mock_response.status_code = 400 mocker.patch("AdminByRequestEventCollector.Client.deny_request", return_value=mock_response) with pytest.raises(DemistoException) as e: deny_request_command(client, args) assert "Failed to deny request 12345. Status code: 400" in str(e.value) def test_deny_request_command_no_id(self, client): """ Given: No request ID. When: Calling the deny_request_command function. Then: Ensure a ValueError is raised. """ with pytest.raises(ValueError) as e: deny_request_command(client, {}) assert "request_id is required" in str(e.value)