import pytest from CommonServerPython import * # noqa # pylint: disable=unused-wildcard-import from pytest_mock import MockerFixture from SymantecEndpointSecurity import ( EventCounter, extract_events_suspected_duplicates, calculate_next_fetch, filter_duplicate_events, perform_long_running_loop, UnauthorizedToken, NextPointingNotAvailable, Client, test_module as _test_module, get_events_command, sleep_if_necessary, ) def mock_client() -> Client: return Client( base_url="test", token="test_token", stream_id="test_stream_id", channel_id="test_channel_id", verify=True, proxy=False, ) @pytest.mark.parametrize( "events, expected_results", [ ( [ {"uuid": "123", "log_time": "2024-10-09T12:34:56Z"}, {"uuid": "456", "log_time": "2024-10-09T12:34:56.789Z"}, {"uuid": "789", "log_time": "2024-10-09T12:34:55.789Z"}, ], (["456"], "2024-10-09T12:34:56.789Z"), ), ( [ {"uuid": "123", "log_time": "2024-10-09T12:34:56Z"}, {"uuid": "456", "log_time": "2024-10-09T12:34:56.78957832Z"}, {"uuid": "789", "log_time": "2024-10-09T12:34:56.789578Z"}, ], (["456", "789"], "2024-10-09T12:34:56.78957832Z"), ), ], ) def test_extract_events_suspected_duplicates(events: list[dict], expected_results: tuple[list[str]]): """ Given - A list of events with timestamps When: - The `extract_events_suspected_duplicates` function is called Then: - Ensure that return a list of UUIDs for events suspected to be duplicates """ assert extract_events_suspected_duplicates(events) == expected_results @pytest.mark.parametrize( "integration_context, events, expected_filtered_events", [ pytest.param( { "events_suspected_duplicates": ["123", "456"], "latest_event_time": "2024-10-09T12:34:56Z", }, [ { "uuid": "123", "log_time": "2024-10-09T12:34:56Z", "time": "2024-10-09T12:34:56Z", }, { "uuid": "456", "log_time": "2024-10-09T12:34:56.789Z", "time": "2024-10-09T12:34:56.789Z", }, { "uuid": "789", "log_time": "2024-10-09T12:34:55.789Z", "time": "2024-10-09T12:34:55.789Z", }, ], [ { "uuid": "456", "log_time": "2024-10-09T12:34:56.789Z", "time": "2024-10-09T12:34:56.789Z", "_time": "2024-10-09T12:34:56.789Z", } ], id="Event time is equal to or less than last_event_time", ), pytest.param( { "events_suspected_duplicates": ["123"], "latest_event_time": "2024-10-09T12:34:56Z", }, [ { "uuid": "123", "log_time": "2024-10-09T12:34:56Z", "time": "2024-10-09T12:34:56Z", }, { "uuid": "456", "log_time": "2024-10-09T12:34:56.789Z", "time": "2024-10-09T12:34:56.789Z", }, ], [ { "uuid": "456", "log_time": "2024-10-09T12:34:56.789Z", "time": "2024-10-09T12:34:56.789Z", "_time": "2024-10-09T12:34:56.789Z", } ], id="Events time is equal to last_event_time but one of them not include in suspected duplicates", ), pytest.param( { "events_suspected_duplicates": ["123"], "latest_event_time": "2024-10-09T12:34:56Z", }, [ { "uuid": "456", "log_time": "2024-10-09T12:35:56.789Z", "time": "2024-10-09T12:35:56.789Z", }, ], [ { "uuid": "456", "log_time": "2024-10-09T12:35:56.789Z", "time": "2024-10-09T12:35:56.789Z", "_time": "2024-10-09T12:35:56.789Z", } ], id="Events time is greater than last_event_time", ), ], ) def test_filter_duplicate_events( integration_context: dict[str, str], events: list[dict[str, str]], expected_filtered_events: list[dict[str, str]], ): """ Given: - A list of events with timestamps When: - The `filter_duplicate_events` function is called Then: - Ensure that a list of the events that are not duplicates is returned """ filtered_events = filter_duplicate_events(events, integration_context, EventCounter()) assert filtered_events == expected_filtered_events @pytest.mark.parametrize( "filtered_events, next_hash, include_last_fetch_events, last_integration_context, expected_integration_context", [ ( [ {"uuid": "12", "log_time": "2024-10-09T12:34:56Z"}, {"uuid": "34", "log_time": "2024-10-09T12:34:56Z"}, {"uuid": "56", "log_time": "2024-10-09T12:34:56Z"}, ], "hash_test_1", False, { "latest_event_time": "2024-10-09T12:34:56Z", "events_suspected_duplicates": ["78", "90"], "next_fetch": {"next": "hash_test"}, "last_fetch_events": [], }, { "latest_event_time": "2024-10-09T12:34:56Z", "events_suspected_duplicates": ["12", "34", "56", "78", "90"], "next_fetch": {"next": "hash_test_1"}, "last_fetch_events": [], }, ) ], ) def test_calculate_next_fetch_last_latest_event_time_are_equal( mocker: MockerFixture, filtered_events: list[dict[str, str]], next_hash: str, include_last_fetch_events: bool, last_integration_context: dict[str, str], expected_integration_context: dict, ): """ Given: - A set of filtered events, next hash, and last integration context When: - The `calculate_next_fetch` function is called Then: - Ensure that updated the 'integration_context' with new events in addition to the old ones, and the next hash """ integration_context = calculate_next_fetch(filtered_events, next_hash, include_last_fetch_events, last_integration_context) assert integration_context == expected_integration_context def test_perform_long_running_loop_unauthorized_token(mocker: MockerFixture): """ Given: - The `perform_long_running_loop` function is called When: - The function is called Then: - Ensure that the function runs indefinitely until the container is stopped """ mocker.patch( "SymantecEndpointSecurity.get_events_command", side_effect=[UnauthorizedToken, Exception("Stop")], ) mock_get_token = mocker.patch.object(Client, "_update_access_token_in_headers") mocker.patch("SymantecEndpointSecurity.sleep_if_necessary") with pytest.raises(DemistoException, match="Failed to fetch logs from API"): perform_long_running_loop(mock_client()) assert mock_get_token.call_count == 2 def test_perform_long_running_loop_next_pointing_not_available(mocker: MockerFixture): """ Given: - No args for the function call When: - The function `perform_long_running_loop` is called Then: - """ mock_integration_context = {"next_fetch": {"next": "test"}} mocker.patch( "SymantecEndpointSecurity.get_events_command", side_effect=[NextPointingNotAvailable, Exception("Stop")], ) mocker.patch.object(Client, "_update_access_token_in_headers") mocker.patch( "SymantecEndpointSecurity.get_integration_context", return_value=mock_integration_context, ) mocker.patch("SymantecEndpointSecurity.sleep_if_necessary") with pytest.raises(DemistoException, match="Failed to fetch logs from API"): perform_long_running_loop(mock_client()) assert mock_integration_context == {"fetch_failure_count": 1} def test_test_module(mocker: MockerFixture): """ Given: - Client When: - The function `test_module` is called Then: - Ensure there is no API call in the test_module function (see the docstring in the `test_module` function). """ mock__http_request = mocker.patch.object(Client, "_http_request") assert _test_module() == "ok" mock__http_request.assert_not_called() @pytest.mark.parametrize( "mock_status_code, exception_type", [ (500, DemistoException), (401, UnauthorizedToken), (410, NextPointingNotAvailable), ], ) def test_get_events_command_with_raises( mocker: MockerFixture, mock_status_code: int, exception_type: type[Exception], ): """ Given: - Client and mock_integration_context When: - The function `get_events_command` is called Then: - Ensure that the function raises Exception based on the status code that returned from the API call """ class MockException: status_code = mock_status_code mocker.patch.object(Client, "_update_access_token_in_headers") mocker.patch.object(Client, "get_events", side_effect=DemistoException("Test", res=MockException())) with pytest.raises(exception_type): get_events_command(mock_client(), {"next_fetch": {"next": "test"}}) @pytest.mark.parametrize( "start_run, end_run, call_count", [ pytest.param(10, 20, 1, id="The sleep function should be called once"), pytest.param(10, 70, 0, id="The sleep function should not be called"), ], ) def test_sleep_if_necessary(mocker: MockerFixture, start_run: int, end_run: int, call_count: int): """ Given: - Mocked time passed duration When: - The function is called Then: - Ensure that the sleep function is called with the appropriate interval value or not called at all if unnecessary. """ mock_sleep = mocker.patch("SymantecEndpointSecurity.time.sleep") sleep_if_necessary(end_run - start_run) assert mock_sleep.call_count == call_count def test_event_counter_without_missing_schema(mocker: MockerFixture): counter = EventCounter() counter.filtered_events = 2 counter.events = 3 counter.total_bytes = 60 demisto_debug_mock = mocker.patch.object(demisto, "debug") counter.print_summary() demisto_debug_mock.assert_called_with( "Summary Log:\n" "- Total events received from Symantec (before filtering): 3 events\n" "- Total events sent to XSIAM (after filtering): 2 events\n" "- Total data received from Symantec: " "60 bytes (~0.0001 MB)\n" "- Number of events missing a schema: 0\n" ) def test_event_counter_with_missing_schema(mocker: MockerFixture): counter = EventCounter() counter.filtered_events = 2 counter.events = 3 counter.total_bytes = 60 counter.event_missing_schema = {"test": "test"} counter.events_missing_schema_counter = 1 demisto_debug_mock = mocker.patch.object(demisto, "debug") counter.print_summary() demisto_debug_mock.assert_called_with("Example of an event missing a schema: {'test': 'test'}")