Code42 Event Collector

Code42 Insider Risk software solutions provide the right balance of transparency, technology and training to detect and appropriately respond to data risk. Use the Code42EventCollector integration to fetch file events and audit logs.

Data Enrichment & Threat Intelligence · Code42

Details

ID	`Code42 Event Collector`
Provider	Permira
Category	Data Enrichment & Threat Intelligence
From Version	8.4.0
Docker Image	`demisto/py42:1.0.0.10120494`
Supported Modules	Agentix XSIAM EDR Cortex Cloud Cloud Runtime Security

README

This is the default integration for this content pack when configured by the Data Onboarder in Cortex XSIAM.

Configure Code42 Event Collector in Cortex

Parameter	Required
Server URL (e.g., https://api.us.code42.com, see help section)	True
API Client ID	True
API Client Secret	True
Maximum number of file events per fetch	True
Maximum number of audit events per fetch	True
Trust any certificate (not secure)	False

Code42 Event Collector Authentication

Code42 API uses the OAuth 2.0 protocol for authentication and authorization.

The domain used for making API requests can be determined using the domain you use to log in to the Code42 console.

Console Domain	API Domain
console.us.code42.com	api.us.code42.com
console.us2.code42.com	api.us2.code42.com
console.ie.code42.com	api.ie.code42.com
console.gov.code42.com	api.gov.code42.com

For each request sent to the API, a bearer token will be requested to authenticate your action. The bearer token should be renewed each 15 minutes. This is done automatically by the integration.

You can retrieve your API credentials by following the instructions in the Code 42 documentation.

Code42 Event Collector Rate Limits

The Code42 API can handle up to 120 requests per minute. After that the API will start to decline client’s requests.

The integration with the default configuration should not raise any rate-limits.

Code42 Event Collector Required Scopes

To use the Code42 Event Collector, make sure you have the correct product plan which must include full Code42 API access.

Commands

You can execute these commands from the CLI, as part of an automation, or in a playbook.
After you successfully execute a command, a DBot message appears in the War Room with the command details.

code42-get-events

Manual command to get events, used mainly for debugging

Base Command

code42-get-events

Input

Argument Name	Description	Required
start_date	Starting time from which to get events.	Required
end_date	Time until when to get events.	Required
limit	The maximum number of events to return. Default is 100.	Required
event_type	The type of event to return. Possible values are: audit-logs, file-events.	Required

Context Output

There is no context output for this command.

Configuration parameters

url — Server URL (e.g., https://api.us.code42.com, see help section) (required)
credentials — API Client ID (required)
max_file_events_per_fetch — Maximum number of file events per fetch
max_audit_events_per_fetch — Maximum number of audit events per fetch
insecure — Trust any certificate (not secure)
event_types_to_fetch — Event Types To Fetch

Commands (1)

code42-get-events
Manual command to get events, used mainly for debugging.

Source

Code42EventCollector.py Code42EventCollector_test.py Code42EventCollector.yml Code42EventCollector_description.md command_examples.txt

Download History Raw

import json
import pytest
from unittest.mock import MagicMock
from freezegun import freeze_time
from pytest_mock import MockerFixture

import requests_toolbelt.sessions

from CommonServerPython import *
from Code42EventCollector import (
    DATE_FORMAT,
    MAX_FETCH_AUDIT_LOGS,
    MAX_FETCH_FILE_EVENTS,
    MAX_AUDIT_LOGS_BATCH_SIZE,
    MAX_FILE_EVENTS_BATCH_SIZE,
    NEXT_TRIGGER_VALUE,
    AuditLogLastRun,
    FileEventLastRun,
)

TEST_URL = "https://test.com"


def create_mocked_response(response: List[Dict] | Dict, status_code: int = 200) -> requests.Response:
    mocked_response = requests.Response()
    mocked_response._content = json.dumps(response).encode("utf-8")
    mocked_response.status_code = status_code
    return mocked_response


def create_file_events(start_id: int, start_date: str, num_of_file_events: int) -> List[Dict[str, Any]]:
    return [
        {"event": {"id": f"{i}", "inserted": (dateparser.parse(start_date) + timedelta(seconds=i)).strftime(DATE_FORMAT)}}
        for i in range(start_id, start_id + num_of_file_events)
    ]


def create_audit_logs(start_id: int, start_date: str, num_of_audit_logs: int) -> List[Dict[str, Any]]:
    return [
        {"id": f"{i}", "timestamp": (dateparser.parse(start_date) + timedelta(seconds=i)).strftime(DATE_FORMAT)}
        for i in range(start_id, start_id + num_of_audit_logs)
    ]


def get_mock_http_request_from_datetimes(datetimes: list[str]):
    def mock_request(method: str, url: str, *args, **kwargs):
        if method == "POST" and "v1/oauth" in url:
            return create_mocked_response(response={"access_token": "1234", "token_type": "bearer", "expires_in": 10000000})
        return create_mocked_response(
            response={
                "fileEvents": [{"event": {"id": f"{i}", "inserted": dt}} for i, dt in enumerate(datetimes)],
                "totalCount": len(datetimes),
            }
        )

    return mock_request


class HttpRequestsMocker:
    latest_file_event_id = 1
    latest_audit_log_id = 1

    def __init__(self, num_of_audit_logs: int = 0, num_of_file_events: int = 0):
        self.num_of_audit_logs = num_of_audit_logs
        self.num_of_file_events = num_of_file_events
        self.fetched_audit_logs = 0
        self.fetched_file_events = 0

    def valid_http_request_side_effect(self, method: str, url: str, *args, **kwargs):
        if method == "POST" and "v1/oauth" in url:
            return create_mocked_response(response={"access_token": "1234", "token_type": "bearer", "expires_in": 10000000})

        if method == "POST" and "/v1/audit/search-audit-log" in url:
            if self.fetched_audit_logs >= self.num_of_audit_logs:
                return create_mocked_response(response={"events": []})

            audit_logs = create_audit_logs(
                self.latest_audit_log_id,
                start_date=(datetime.utcfromtimestamp(kwargs["json"]["dateRange"]["startTime"])).strftime(DATE_FORMAT),
                num_of_audit_logs=min(kwargs["json"]["pageSize"], self.num_of_audit_logs),
            )

            self.fetched_audit_logs += len(audit_logs)

            self.latest_audit_log_id = int(audit_logs[-1]["id"]) + 1
            return create_mocked_response(response={"events": audit_logs})

        if method == "POST" and "/v2/file-events" in url:
            if self.fetched_file_events >= self.num_of_file_events:
                return create_mocked_response({"fileEvents": []})

            file_events = create_file_events(
                self.latest_file_event_id,
                start_date="2024-01-24 12:30:45.123456Z",
                num_of_file_events=min(kwargs["json"]["pgSize"], self.num_of_file_events),
            )

            self.fetched_file_events += len(file_events)

            self.latest_file_event_id = int(file_events[-1]["event"]["id"]) + 1
            return create_mocked_response(response={"fileEvents": file_events, "totalCount": self.num_of_file_events})
        return None


def test_the_test_module(mocker):
    """
    Given:
     - a single audit log and a single file event
     - api returns 200 ok

    When:
     - running test-module

    Then:
     - make sure the test is successful.
    """
    import Code42EventCollector

    return_results_mocker: MagicMock = mocker.patch.object(Code42EventCollector, "return_results")
    mocker.patch.object(
        demisto,
        "params",
        return_value={
            "url": TEST_URL,
            "credentials": {
                "identifier": "1234",
                "password": "1234",
            },
        },
    )
    mocker.patch.object(demisto, "command", return_value="test-module")

    mocker.patch.object(
        requests_toolbelt.sessions.BaseUrlSession,
        "request",
        side_effect=HttpRequestsMocker(num_of_file_events=1, num_of_audit_logs=1).valid_http_request_side_effect,
    )

    Code42EventCollector.main()
    assert return_results_mocker.called
    assert return_results_mocker.call_args[0][0] == "ok"


def test_fetch_events_no_last_run(mocker):
    """
    Given:
     - a single audit log and a single file event
     - api returns 200 ok

    When:
     - running fetch events

    Then:
     - make sure events are sent successfully
     - make sure last run is populated correctly
    """
    import Code42EventCollector

    send_events_mocker: MagicMock = mocker.patch.object(Code42EventCollector, "send_events_to_xsiam")
    mocker.patch.object(
        demisto,
        "params",
        return_value={
            "url": TEST_URL,
            "credentials": {
                "identifier": "1234",
                "password": "1234",
            },
            "event_types_to_fetch": "File,Audit",
        },
    )
    set_last_run_mocker: MagicMock = mocker.patch.object(demisto, "setLastRun")
    update_module_health_mocker: MagicMock = mocker.patch.object(demisto, "updateModuleHealth")
    mocker.patch.object(demisto, "getLastRun", return_value={})
    mocker.patch.object(demisto, "command", return_value="fetch-events")
    mocker.patch.object(
        requests_toolbelt.sessions.BaseUrlSession,
        "request",
        side_effect=HttpRequestsMocker(num_of_file_events=1, num_of_audit_logs=1).valid_http_request_side_effect,
    )

    Code42EventCollector.main()
    file_events = send_events_mocker.call_args_list[0][0][0]
    assert len(file_events) == 1
    assert file_events[0]["eventType"] == Code42EventCollector.EventType.FILE
    assert update_module_health_mocker.call_args_list[0][0][0] == {"eventsPulled": len(file_events)}

    audit_logs = send_events_mocker.call_args_list[1][0][0]
    assert len(audit_logs) == 1
    assert audit_logs[0]["eventType"] == Code42EventCollector.EventType.AUDIT
    assert update_module_health_mocker.call_args_list[1][0][0] == {"eventsPulled": len(audit_logs)}

    last_run_expected_keys = {
        Code42EventCollector.FileEventLastRun.FETCHED_IDS,
        Code42EventCollector.FileEventLastRun.TIME,
        Code42EventCollector.FileEventLastRun.CUMULATIVE_COUNT,
        Code42EventCollector.FileEventLastRun.NEXT_TRIGGER,
        Code42EventCollector.AuditLogLastRun.FETCHED_IDS,
        Code42EventCollector.AuditLogLastRun.TIME,
        Code42EventCollector.AuditLogLastRun.CUMULATIVE_COUNT,
        Code42EventCollector.AuditLogLastRun.NEXT_TRIGGER,
        "nextTrigger",
    }

    assert last_run_expected_keys == set(set_last_run_mocker.call_args_list[0][0][0].keys())


def test_fetch_events_no_last_run_max_fetch_lower_than_available_events(mocker):
    """
    Given:
     - 550 audit logs and 550 file events
     - api returns 200 ok
     - max fetch = 500

    When:
     - running fetch events

    Then:
     - make sure 500 events are sent successfully
     - make sure last run is populated correctly
    """
    import Code42EventCollector

    send_events_mocker: MagicMock = mocker.patch.object(Code42EventCollector, "send_events_to_xsiam")
    mocker.patch.object(
        demisto,
        "params",
        return_value={
            "url": TEST_URL,
            "credentials": {
                "identifier": "1234",
                "password": "1234",
            },
            "max_file_events_per_fetch": 500,
            "max_audit_events_per_fetch": 500,
            "event_types_to_fetch": "File,Audit",
        },
    )
    set_last_run_mocker: MagicMock = mocker.patch.object(demisto, "setLastRun")
    update_module_health_mocker: MagicMock = mocker.patch.object(demisto, "updateModuleHealth")
    mocker.patch.object(demisto, "getLastRun", return_value={})
    mocker.patch.object(demisto, "command", return_value="fetch-events")
    mocker.patch.object(
        requests_toolbelt.sessions.BaseUrlSession,
        "request",
        side_effect=HttpRequestsMocker(num_of_file_events=550, num_of_audit_logs=550).valid_http_request_side_effect,
    )

    Code42EventCollector.main()
    file_events = send_events_mocker.call_args_list[0][0][0]
    assert len(file_events) == 500
    for file_event in file_events:
        assert file_event["eventType"] == Code42EventCollector.EventType.FILE
    assert update_module_health_mocker.call_args_list[0][0][0] == {"eventsPulled": len(file_events)}

    audit_logs = send_events_mocker.call_args_list[1][0][0]
    assert len(audit_logs) == 500
    for audit_log in audit_logs:
        assert audit_log["eventType"] == Code42EventCollector.EventType.AUDIT
    assert update_module_health_mocker.call_args_list[1][0][0] == {"eventsPulled": len(audit_logs)}

    last_run_expected_keys = {
        Code42EventCollector.FileEventLastRun.FETCHED_IDS,
        Code42EventCollector.FileEventLastRun.TIME,
        Code42EventCollector.AuditLogLastRun.FETCHED_IDS,
        Code42EventCollector.AuditLogLastRun.TIME,
    }

    # make sure all keys in last run are valid
    assert last_run_expected_keys.issubset(set(set_last_run_mocker.call_args_list[0][0][0].keys()))


def test_fetch_events_no_last_run_no_audit_logs_yes_file_events(mocker):
    """
    Given:
     - 0 audit logs and 100 file events
     - api returns 200 ok
     - max fetch = 500

    When:
     - running fetch events

    Then:
     - make sure 100 file events are sent successfully
     - make sure no audit logs are sent
     - make sure last run is populated correctly
    """
    import Code42EventCollector

    send_events_mocker: MagicMock = mocker.patch.object(Code42EventCollector, "send_events_to_xsiam")
    mocker.patch.object(
        demisto,
        "params",
        return_value={
            "url": TEST_URL,
            "credentials": {
                "identifier": "1234",
                "password": "1234",
            },
            "max_file_events_per_fetch": 500,
            "max_audit_events_per_fetch": 500,
            "event_types_to_fetch": "File,Audit",
        },
    )
    set_last_run_mocker: MagicMock = mocker.patch.object(demisto, "setLastRun")
    update_module_health_mocker: MagicMock = mocker.patch.object(demisto, "updateModuleHealth")
    mocker.patch.object(demisto, "getLastRun", return_value={})
    mocker.patch.object(demisto, "command", return_value="fetch-events")
    mocker.patch.object(
        requests_toolbelt.sessions.BaseUrlSession,
        "request",
        side_effect=HttpRequestsMocker(num_of_file_events=100, num_of_audit_logs=0).valid_http_request_side_effect,
    )

    Code42EventCollector.main()
    file_events = send_events_mocker.call_args_list[0][0][0]
    assert len(file_events) == 100
    for file_event in file_events:
        assert file_event["eventType"] == Code42EventCollector.EventType.FILE
    assert update_module_health_mocker.call_args_list[0][0][0] == {"eventsPulled": len(file_events)}

    audit_logs = send_events_mocker.call_args_list[1][0][0]
    assert len(audit_logs) == 0
    assert update_module_health_mocker.call_args_list[1][0][0] == {"eventsPulled": len(audit_logs)}

    last_run_expected_keys = {
        Code42EventCollector.FileEventLastRun.FETCHED_IDS,
        Code42EventCollector.FileEventLastRun.TIME,
        Code42EventCollector.FileEventLastRun.CUMULATIVE_COUNT,
        Code42EventCollector.FileEventLastRun.NEXT_TRIGGER,
        Code42EventCollector.AuditLogLastRun.CUMULATIVE_COUNT,
        Code42EventCollector.AuditLogLastRun.NEXT_TRIGGER,
        "nextTrigger",
    }

    assert last_run_expected_keys == set(set_last_run_mocker.call_args_list[0][0][0].keys())


def test_fetch_events_no_last_run_yes_audit_logs_no_file_events(mocker):
    """
    Given:
     - 100 audit logs and 0 file events
     - api returns 200 ok
     - max fetch = 500

    When:
     - running fetch events

    Then:
     - make sure 100 audit logs are sent successfully
     - make sure no file events are sent
     - make sure last run is populated correctly
    """
    import Code42EventCollector

    send_events_mocker: MagicMock = mocker.patch.object(Code42EventCollector, "send_events_to_xsiam")
    mocker.patch.object(
        demisto,
        "params",
        return_value={
            "url": TEST_URL,
            "credentials": {
                "identifier": "1234",
                "password": "1234",
            },
            "max_file_events_per_fetch": 500,
            "max_audit_events_per_fetch": 500,
            "event_types_to_fetch": "File,Audit",
        },
    )
    set_last_run_mocker: MagicMock = mocker.patch.object(demisto, "setLastRun")
    mocker.patch.object(demisto, "getLastRun", return_value={})
    mocker.patch.object(demisto, "command", return_value="fetch-events")
    mocker.patch.object(
        requests_toolbelt.sessions.BaseUrlSession,
        "request",
        side_effect=HttpRequestsMocker(num_of_file_events=0, num_of_audit_logs=100).valid_http_request_side_effect,
    )

    Code42EventCollector.main()
    file_events = send_events_mocker.call_args_list[0][0][0]
    assert len(file_events) == 0

    audit_logs = send_events_mocker.call_args_list[1][0][0]
    assert len(audit_logs) == 100
    for audit_log in audit_logs:
        assert audit_log["eventType"] == Code42EventCollector.EventType.AUDIT

    last_run_expected_keys = {
        Code42EventCollector.AuditLogLastRun.TIME,
        Code42EventCollector.AuditLogLastRun.FETCHED_IDS,
        Code42EventCollector.AuditLogLastRun.CUMULATIVE_COUNT,
        Code42EventCollector.AuditLogLastRun.NEXT_TRIGGER,
        Code42EventCollector.FileEventLastRun.NEXT_TRIGGER,
        Code42EventCollector.FileEventLastRun.CUMULATIVE_COUNT,
        "nextTrigger",
    }

    assert last_run_expected_keys == set(set_last_run_mocker.call_args_list[0][0][0].keys())


def test_fetch_events_no_last_run_no_events(mocker):
    """
    Given:
     - no audit logs | no file events
     - api returns 200 ok

    When:
     - running fetch events

    Then:
     - make sure no events were sent
    """
    import Code42EventCollector

    send_events_mocker: MagicMock = mocker.patch.object(Code42EventCollector, "send_events_to_xsiam")
    mocker.patch.object(
        demisto,
        "params",
        return_value={
            "url": TEST_URL,
            "credentials": {
                "identifier": "1234",
                "password": "1234",
            },
            "event_types_to_fetch": "File,Audit",
        },
    )
    mocker.patch.object(demisto, "setLastRun")
    mocker.patch.object(demisto, "getLastRun", return_value={})
    mocker.patch.object(demisto, "command", return_value="fetch-events")
    mocker.patch.object(
        requests_toolbelt.sessions.BaseUrlSession,
        "request",
        side_effect=HttpRequestsMocker(num_of_file_events=0, num_of_audit_logs=0).valid_http_request_side_effect,
    )

    Code42EventCollector.main()

    file_events = send_events_mocker.call_args_list[0][0][0]
    assert len(file_events) == 0

    audit_logs = send_events_mocker.call_args_list[1][0][0]
    assert len(audit_logs) == 0


@freeze_time("2024-01-01 01:00:15 UTC")
def test_fetch_events_within_look_back(mocker: MockerFixture):
    """
    Given:
     - A run with incidents within the look-back time frame.

    When:
     - Running fetch events.

    Then:
     - The next-fetch should be the look-back time rounded down to the first three microsecond digits,
       and all incidents later than the next-fetch should be kept in the last-run.
    """
    from Code42EventCollector import main, FILE_EVENTS_LOOK_BACK, FileEventLastRun

    LOOK_BACK_TIME = datetime.now() - FILE_EVENTS_LOOK_BACK

    mocker.patch("Code42EventCollector.send_events_to_xsiam")
    mocker.patch.object(
        demisto,
        "params",
        return_value={
            "url": TEST_URL,
            "credentials": {
                "identifier": "1234",
                "password": "1234",
            },
            "max_file_events_per_fetch": 500,
            "max_audit_events_per_fetch": 500,
            "event_types_to_fetch": "File",
        },
    )
    mocker.patch.object(
        requests_toolbelt.sessions.BaseUrlSession,
        "request",
        side_effect=get_mock_http_request_from_datetimes(
            [
                (LOOK_BACK_TIME + timedelta(seconds=-1)).strftime(DATE_FORMAT),
                (LOOK_BACK_TIME + timedelta(microseconds=500_200)).strftime(DATE_FORMAT),
                (LOOK_BACK_TIME + timedelta(microseconds=500_100)).strftime(DATE_FORMAT),
                (LOOK_BACK_TIME + timedelta(microseconds=500_300)).strftime(DATE_FORMAT),
            ]
        ),
    )
    set_last_run_mock = mocker.patch.object(demisto, "setLastRun")
    mocker.patch.object(demisto, "command", return_value="fetch-events")

    main()

    assert set(set_last_run_mock.call_args_list[0][0][0][FileEventLastRun.FETCHED_IDS]) == {"1", "2", "3"}
    assert set_last_run_mock.call_args_list[0][0][0][FileEventLastRun.TIME] == "2024-01-01 00:59:30.000000Z"


@freeze_time("2024-01-01 01:00:15 UTC")
def test_fetch_events_before_look_back(mocker: MockerFixture):
    """
    Given:
     - A run with incidents before the look-back time frame.

    When:
     - Running fetch events.

    Then:
     - The next-fetch should be the latest creation time rounded down to the first three microsecond digits,
       and all incidents later than the next-fetch should be kept in the last-run.
    """
    from Code42EventCollector import main, FILE_EVENTS_LOOK_BACK, FileEventLastRun

    LOOK_BACK_TIME = datetime.now() - FILE_EVENTS_LOOK_BACK

    mocker.patch("Code42EventCollector.send_events_to_xsiam")
    mocker.patch.object(
        demisto,
        "params",
        return_value={
            "url": TEST_URL,
            "credentials": {
                "identifier": "1234",
                "password": "1234",
            },
            "max_file_events_per_fetch": 500,
            "max_audit_events_per_fetch": 500,
            "event_types_to_fetch": "File",
        },
    )
    mocker.patch.object(
        requests_toolbelt.sessions.BaseUrlSession,
        "request",
        side_effect=get_mock_http_request_from_datetimes(
            [
                (LOOK_BACK_TIME + timedelta(minutes=-1, microseconds=100_000)).strftime(DATE_FORMAT),
                (LOOK_BACK_TIME + timedelta(minutes=-1, microseconds=500_200)).strftime(DATE_FORMAT),
                (LOOK_BACK_TIME + timedelta(minutes=-1, microseconds=500_100)).strftime(DATE_FORMAT),
                (LOOK_BACK_TIME + timedelta(minutes=-1, microseconds=500_300)).strftime(DATE_FORMAT),
            ]
        ),
    )
    set_last_run_mock = mocker.patch.object(demisto, "setLastRun")
    mocker.patch.object(demisto, "command", return_value="fetch-events")

    main()

    assert set(set_last_run_mock.call_args_list[0][0][0][FileEventLastRun.FETCHED_IDS]) == {"1", "2", "3"}
    assert set_last_run_mock.call_args_list[0][0][0][FileEventLastRun.TIME] == "2024-01-01 00:58:30.500000Z"


@freeze_time("2024-01-01 01:00:15 UTC")
def test_fetch_events_empty_run(mocker: MockerFixture):
    """
    Given:
     - An empty run.

    When:
     - Running fetch events.

    Then:
     - The last-run should stay the same.
    """
    from Code42EventCollector import main

    mocker.patch("Code42EventCollector.send_events_to_xsiam")
    mocker.patch.object(
        demisto,
        "params",
        return_value={
            "url": TEST_URL,
            "credentials": {
                "identifier": "1234",
                "password": "1234",
            },
            "max_file_events_per_fetch": 500,
            "max_audit_events_per_fetch": 500,
            "event_types_to_fetch": "File",
        },
    )
    mocker.patch.object(
        requests_toolbelt.sessions.BaseUrlSession, "request", side_effect=HttpRequestsMocker().valid_http_request_side_effect
    )
    mocker.patch.object(demisto, "getLastRun", return_value={"LastRun": "previous_last_run"})
    set_last_run_mock = mocker.patch.object(demisto, "setLastRun")
    mocker.patch.object(demisto, "command", return_value="fetch-events")

    main()

    assert set_last_run_mock.call_args_list[0][0][0] == {
        "LastRun": "previous_last_run",
        "file-event-count": 0,
        "file-event-next-trigger": None,
        "nextTrigger": None,
    }


def test_get_events_command(mocker):
    """
    Given:
     - 1 audit log / 1 file event
     - api returns 200 ok

    When:
     - running get_events_command

    Then:
     - make sure the events are returned as expected
    """
    import Code42EventCollector

    return_results_mocker: MagicMock = mocker.patch.object(Code42EventCollector, "return_results")

    mocker.patch.object(
        demisto,
        "params",
        return_value={
            "url": TEST_URL,
            "credentials": {
                "identifier": "1234",
                "password": "1234",
            },
        },
    )

    mocker.patch.object(
        demisto, "args", return_value={"start_date": datetime.utcnow() - timedelta(minutes=1), "event_type": "audit"}
    )

    mocker.patch.object(demisto, "command", return_value="code42-get-events")
    mocker.patch.object(
        requests_toolbelt.sessions.BaseUrlSession,
        "request",
        side_effect=HttpRequestsMocker(num_of_file_events=1, num_of_audit_logs=1).valid_http_request_side_effect,
    )

    Code42EventCollector.main()

    command_result = return_results_mocker.call_args_list[0][0][0]
    assert command_result.outputs[0]["eventType"] == Code42EventCollector.EventType.AUDIT
    assert len(command_result.outputs) == 1
    assert command_result.outputs
    assert command_result.readable_output

    mocker.patch.object(
        demisto, "args", return_value={"start_date": datetime.utcnow() - timedelta(minutes=1), "event_type": "file"}
    )

    Code42EventCollector.main()

    command_result = return_results_mocker.call_args_list[1][0][0]
    assert len(command_result.outputs) == 1
    assert command_result.outputs[0]["eventType"] == Code42EventCollector.EventType.FILE
    assert command_result.outputs
    assert command_result.readable_output


@pytest.mark.parametrize(
    "last_run, expected_files_count, expected_files_next_trigger, expected_audits_count, expected_audits_next_trigger",
    [
        pytest.param(  # Empty last run (counts incremented and nextTrigger in 3 seconds)
            {},
            MAX_FILE_EVENTS_BATCH_SIZE,
            NEXT_TRIGGER_VALUE,
            MAX_AUDIT_LOGS_BATCH_SIZE,
            NEXT_TRIGGER_VALUE,
            id="First fetch",
        ),
        pytest.param(  # Halfway through batching (counts incremented and nextTrigger in 3 seconds)
            {FileEventLastRun.CUMULATIVE_COUNT.value: 12000, AuditLogLastRun.CUMULATIVE_COUNT.value: 48000},
            12000 + MAX_AUDIT_LOGS_BATCH_SIZE,
            NEXT_TRIGGER_VALUE,
            48000 + MAX_AUDIT_LOGS_BATCH_SIZE,
            NEXT_TRIGGER_VALUE,
            id="Middle batch",
        ),
        pytest.param(  # Last batch before reaching max fetch (counts reset and no nextTrigger)
            {FileEventLastRun.CUMULATIVE_COUNT.value: 48000, AuditLogLastRun.CUMULATIVE_COUNT.value: 96000},
            0,
            None,
            0,
            None,
            id="Last batch",
        ),
    ],
)
def test_next_trigger(
    last_run: dict,
    expected_files_count: int,
    expected_files_next_trigger: str | None,
    expected_audits_count: int,
    expected_audits_next_trigger: str | None,
):
    """
    Given:
     - last run value and "File" and "Audit" event types to fetch.

    When:
     - Fetching in batches.

    Then:
     - Assert the cumulative count is correctly incremented and nextTrigger is set in the next run when batching is in progress.
     - Assert the cumulative count is reset and nextTrigger is None in the next run when batching completes.
    """
    from Code42EventCollector import fetch_audit_logs, fetch_file_events

    class MockClient:
        def get_file_events(*_, **__) -> list[dict]:
            return [
                {"event": {"id": f"{i}"}, "_time": datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(seconds=i)}
                for i in range(MAX_FILE_EVENTS_BATCH_SIZE)
            ]

        def get_audit_logs(*_, **__) -> list[dict]:
            return [
                {"event": {"id": f"{j}"}, "_time": datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(seconds=j)}
                for j in range(MAX_AUDIT_LOGS_BATCH_SIZE)
            ]

    _, file_event_last_run = fetch_file_events(MockClient, last_run, MAX_FETCH_FILE_EVENTS)  # type: ignore
    _, audit_logs_last_run = fetch_audit_logs(MockClient, last_run, MAX_FETCH_AUDIT_LOGS)  # type: ignore

    assert file_event_last_run[FileEventLastRun.CUMULATIVE_COUNT.value] == expected_files_count
    assert file_event_last_run[FileEventLastRun.NEXT_TRIGGER.value] == expected_files_next_trigger

    assert audit_logs_last_run[AuditLogLastRun.CUMULATIVE_COUNT.value] == expected_audits_count
    assert audit_logs_last_run[AuditLogLastRun.NEXT_TRIGGER.value] == expected_audits_next_trigger