GenericAPIEventCollector

Collect logs from 3rd party vendors using API.

Analytics & SIEM · GenericAPIEventCollector

Details

ID	`GenericAPIEventCollector`
Provider	Open Source
Category	Analytics & SIEM
From Version	6.8.0
Docker Image	`demisto/python3:3.12.13.10116658`
Supported Modules	XSIAM

README

The Generic API Event Collector allows you to ingest data from any API endpoint into Cortex.
By configuring this collector, you can gather data from various systems and bring it into the Cortex ecosystem for better analysis and correlation.

Note: This pack is currently in Beta, and as such, it may be subject to future changes and may not work on all types of APIs and Authentication.

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

Configure Generic API Event Collector (Beta) in Cortex

Parameter	Description	Required
Server URL		True
Endpoint	Add the endpoint you want to collect data from (Alert/Events etc.).	True
Authentication Type	Select the authentication method.	True
HTTP Method	The HTTP method of the request to the API.	True
API Token	API Key to access the service REST API.	False
Username	Username & Password to use for basic authentication.	False
Password		False
Add Fields To header	If the product authentication requires more fields to add to the header add it here in dictionary format {unique_field : 286}. If there’s a need to add more then one, use it in this format: {‘field-1’: value_example, ‘field-2’: value_2, ‘field-3’: value_3}	False
Use system proxy settings		False
Trust any certificate (not secure)		False
Vendor	Enter vendor name for dataset.	True
Product	Enter product name for dataset.	True
Request data	If the product authentication requires more fields to add to the DATA add it here in dictionary format {unique_field : 286}. If there’s a need to add more then one, use it in this format: {‘field-1’: value_example, ‘field-2’: value_2, ‘field-3’: value_3}	False
Initial request data	If the product requires a different initial DATA, add it here in dictionary format {unique_field : 286}. If there’s a need to add more then one, use it in this format: {‘field-1’: value_example, ‘field-2’: value_2, ‘field-3’: value_3}	False
Request JSON parameters	If the product authentication requires more fields to add to the body as JSON add it here in dictionary format {unique_field : 286}. If there’s a need to add more then one, use it in this format: {‘field-1’: value_example, ‘field-2’: value_2, ‘field-3’: value_3}	False
Initial request JSON parameters	If the product requires a different initial request JSON, add it here in dictionary format {unique_field : 286}. If there’s a need to add more then one, use it in this format: {‘field-1’: value_example, ‘field-2’: value_2, ‘field-3’: value_3}	False
Query parameters	If the product authentication allows to filter the results using query Parameters add it here in dictionary format {unique_field : 286}. If there’s a need to add more then one, use it in this format: {‘field-1’: value_example, ‘field-2’: value_2, ‘field-3’: value_3}	False
Initial query parameters	If the product requires a different initial query parameters for the first fetch call, add it here in dictionary format {unique_field : 286}. If there’s a need to add more then one, use it in this format: {‘field-1’: value_example, ‘field-2’: value_2, ‘field-3’: value_3}	False
Is pagination needed	If the API JSON response supports events pagination.
Pagination field name	Next page field in JSON response, e.g., “cursor”, “next_page”	False
Pagination flag	Next page existence in JSON response e.g., “has_more”, “next”	False
Timestamp format of the event creation time or “epoch”.	Python compatible datetime formatting (e.g. ,”%Y-%m-%dT%H:%M:%S.%fZ” or “%Y.%m.%d %H:%M:%S”) or “epoch” to use UNIX epoch time.	False
Timestamp field	The name of the event creation time in the response data, e.g., “timestamp” or “created_at”.	True
Events lookup path in the response JSON, dot-separated, e.g., “data.items”.	Where within the response object to find the events list.	False
Event ID lookup path in the event response JSON, dot-separated, e.g., “id”.	Where within the event object to find the event ID.	False
The type of ID field, either “integer” or “string”	ID field of type integer are comparable and when last fetched ID is the maximum ID between the fetched events, when the type is string, the last fetched ID is the last event returned from the API.	False
OK codes	Allowed HTTP status codes for successful response from the API	False
Limit	Number of incidents to fetch per fetch.	False
Fetch Events		False
Events Fetch Interval		False

How to configure the event collector

Authentication

You must specify the authentication method required by the server.
The supported authentication types include:

Basic authentication (username and password)
Token-based authentication
Bearer token
Api-Key token
Raw Token (for custom token-based authentication)
No Authorization (for publicly accessible data)

Pagination

When the API supports pagination in the response, the collector can fetch more pages of data using the following parameters:

Is pagination needed If the API JSON response supports events pagination.
Pagination field name, Next page field in JSON response, e.g., “cursor” or “next_page” False
Pagination flag, The Next page existence in JSON response e.g., “has_more” or “next”

In the below example the pagination flag is pagination.has_more
The pagination field name is pagination.next_page

{
    "data": [
        {
            "id": 1,
            "name": "John Doe",
            "occupation": "Software Engineer"
        },
        {
            "id": 2,
            "name": "Jane Smith",
            "occupation": "Data Scientist"
        }
    ],
    "pagination": {
        "current_page": 1,
        "next_page": "https://api.example.com/users?page=2",
        "has_more": true
    }
}

Request Data (And initial request data)

If the product authentication requires more fields to add to the DATA.
Add it here in dictionary format.

For example:

{"field-1": "value_example", "field-2": 1, "field-3": "value_3"}

Note: Using the initial request data parameter will only be used in the first request to collect events.

Request JSON (And initial request JSON)

If the product authentication requires more fields to add to the body as JSON, add it
here in dictionary format.

For example:

{"date": "2021-08-01", "field-2": 1, "field-3": "value_3"}

Note: Using the initial request JSON parameter will only be used in the first request to collect events.

Query parameters (And Initial Query parameters)

If the product authentication allows filtering the results using query parameters, add it here in dictionary format:

{"ordering": "id", "limit": 1, "created_after": "@first_fetch_datetime"}

Note: Using the initial query parameters parameter will only be used in the first request to collect events.

Timestamp field

The name of the event creation time in the response data, e.g., “timestamp” or “created_at”.
In the following API response:

{
  "data": [
    {
      "id": 3,
      "name": "Alice Brown",
      "occupation": "Network Engineer",
      "created": "2021-10-05T19:45:20.789012Z"
    },
    {
      "id": 4,
      "name": "Dave Testson",
      "occupation": "Cybersecurity Analyst",
      "created": "2021-10-06T10:15:45.654321Z"
    }
  ],
  "pagination": {
    "current_page": 2,
    "next_page": "https://api.example.com/users?page=3",
    "has_more": true
  }
}

the timestamp field is created

Timestamp format

The timestamp format of the event creation time or “epoch” to use UNIX epoch time.
The formatting supported is Python-compatible datetime formatting (e.g., “%Y-%m-%dT%H:%M:%S.%fZ” or “%Y.%m.%d %H:%M:%S”).
In the following API response:

{
  "data": [
    {
      "id": 3,
      "name": "Alice Brown",
      "occupation": "Network Engineer",
      "created": "2021-10-05T19:45:20.789012Z"
    },
    {
      "id": 4,
      "name": "Dave Testson",
      "occupation": "Cybersecurity Analyst",
      "created": "2021-10-06T10:15:45.654321Z"
    }
  ],
  "pagination": {
    "current_page": 2,
    "next_page": "https://api.example.com/users?page=3",
    "has_more": true
  }
}

The timestamp format is python format “%Y-%m-%dT%H:%M:%S.%fZ”

Note: To learn more about Python date and time formats, see: https://docs.python.org/3/library/datetime.html#format-codes

Events

Where within the response JSON to search for the events, dot-separated (e.g., “data.items”).

Example 1:

{
  "data": [
    {
      "id": 4,
      "name": "Alice Brown",
      "occupation": "Network Engineer",
      "created": "2021-10-05T19:45:20.789012Z"
    },
    {
      "id": 3,
      "name": "Dave Testson",
      "occupation": "Cybersecurity Analyst",
      "created": "2021-10-06T10:15:45.654321Z"
    }
  ]
}

The events are within the “data” in the response.

Example 2:

{
  "data": {
    "items": [
        {
            "id": 4,
            "name": "Alice Brown",
            "occupation": "Network Engineer",
            "created": "2021-10-05T19:45:20.789012Z"
        },
        {
            "id": 3,
            "name": "Dave Testson",
            "occupation": "Cybersecurity Analyst",
            "created": "2021-10-06T10:15:45.654321Z"
        }
    ]
  }
}

The events are within the “data.items” in the response.

Event ID & Type

Event ID lookup path in the event response JSON, dot-separated, e.g., “id”
Where within the event object to find the event ID.

The type of ID field, either “integer” or “string”:

ID field of type integer is comparable, and when last fetched ID is the maximum ID between the fetched events.
ID field of the type is string, the last fetched ID is the last event returned from the API.

Example 1:

{
  "data": [
    {
      "id": 4,
      "name": "Alice Brown",
      "occupation": "Network Engineer",
      "created": "2021-10-05T19:45:20.789012Z"
    },
    {
      "id": 3,
      "name": "Dave Testson",
      "occupation": "Cybersecurity Analyst",
      "created": "2021-10-06T10:15:45.654321Z"
    }
  ],
  "pagination": {
    "current_page": 2,
    "next_page": "https://api.example.com/users?page=3",
    "has_more": true
  }
}

The event ID field should be “id” and the type should be integer, and the last fetched ID will be 4.

Example 2:

{
  "data": [
    {
      "uuid": "123e4567-e89b-12d3-a456-426614174000",
      "name": "Alice Brown",
      "occupation": "Network Engineer",
      "created": "2021-10-05T19:45:20.789012Z"
    },
    {
      "uuid": "123e4567-e89b-12d3-a456-426614174001",
      "name": "Dave Testson",
      "occupation": "Cybersecurity Analyst",
      "created": "2021-10-06T10:15:45.654321Z"
    }
  ],
  "pagination": {
    "current_page": 2,
    "next_page": "https://api.example.com/users?page=3",
    "has_more": true
  }
}

The event ID field should be “uuid” and the type should be string, and the last fetched ID will be “123e4567-e89b-12d3-a456-426614174001”.

Substitutions in API requests calls

To make the API calls more dynamic against the API endpoint, we added a few placeholders that will be substituted before calling the API endpoint.

@last_fetched_id - The last ID that was fetched from the API, if this is the first fetch, the value will be empty.
@last_fetched_datetime - The last fetched event time from the API, if this is the first fetch, the value will be empty.
@first_fetch_datetime - The first fetch time, when the integration first started to fetch events.
@fetch_size_limit - The number of incidents to fetch per fetch.

Examples being used in query parameters:

This will substitute the @last_fetched_id with the last fetched ID from a previous fetch call.

{"ordering": "id", "limit": 100, "id__gt": "@last_fetched_id"}

The resulting API query parameters will be:

{"ordering": "id", "limit": 100, "id__gt": "4"}

This will substitute the @first_fetch_datetime with the first fetch time.

{"ordering": "id", "limit": 1, "created_after": "@first_fetch_datetime"}

The resulting API query parameters will be:

{"ordering": "id", "limit": 1, "created_after": "2021-10-06T10:15:45.654321Z"}

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.

generic-api-event-collector-get-events

Gets events from 3rd-party vendor.

Base Command

generic-api-event-collector-get-events

Input

Argument Name	Description	Required
should_push_events	If true, the command will create events, otherwise it will only display them. Possible values are: true, false. Default is false.	Required
limit	Maximum number of results to return.	Optional

Context Output

There is no context output for this command.

Configuration parameters

base_url — Server URL (required)
endpoint — Endpoint (required)
authentication — Authentication Type (required)
http_method — HTTP Method (required)
token —
credentials — Username
add_fields_to_header — Add Fields To header
proxy — Use system proxy settings
insecure — Trust any certificate (not secure)
vendor — Vendor (required)
product — Product (required)
request_data — Request data
initial_request_data — Initial request data
request_json — Request JSON parameters
initial_request_json — Initial request JSON parameters
query_params — Query parameters
initial_query_params — Initial query parameters
pagination_needed — Is pagination needed
pagination_field_name — Pagination field name
pagination_flag — Pagination flag
timestamp_format — Timestamp format of the event creation time or "epoch".
timestamp_field_name — Timestamp field (required)
events_keys — Events lookup path in the response JSON, dot-separated, e.g. ,"data.items".
id_keys — Event ID lookup path in the event response JSON, dot-separated, e.g., "id".
id_type — The type of ID field, either "integer" or "string"
ok_codes — Allowed HTTP status codes for successful response from the API
limit — Number of incidents to fetch per fetch.
isFetchEvents — Fetch Events
eventFetchInterval — Events Fetch Interval

Commands (1)

generic-api-event-collector-get-events
Gets events from 3rd party vendor.

Source

GenericAPIEventCollector.py GenericAPIEventCollector_test.py GenericAPIEventCollector.yml GenericAPIEventCollector_description.md command_examples.txt

Download History Raw

from datetime import datetime
from unittest.mock import patch

import pytest

from CommonServerPython import DemistoException
from GenericAPIEventCollector import (
    PaginationLogic,
    RequestData,
    TimestampFieldConfig,
    datetime_to_timestamp_format,
    extract_pagination_params,
    generate_authentication_headers,
    generate_headers,
    get_time_field_from_event_to_dt,
    is_pagination_needed,
    iso8601_to_datetime_str,
    organize_events_to_xsiam_format,
    parse_json_param,
    recursive_replace,
    setup_search_events,
    timestamp_format_to_datetime,
    is_milliseconds,
    is_microseconds,
    convert_epoch_to_timestamp,
)


def test_datetime_to_timestamp_format():
    dt = datetime(2023, 5, 1, 12, 0, 0)
    assert datetime_to_timestamp_format(dt, "%Y-%m-%dT%H:%M:%SZ") == "2023-05-01T12:00:00Z"
    assert datetime_to_timestamp_format(dt, "epoch") == str(dt.timestamp())


def test_timestamp_format_to_datetime():
    dt_str = "2023-05-01T12:00:00Z"
    assert timestamp_format_to_datetime(dt_str, "%Y-%m-%dT%H:%M:%SZ") == datetime(2023, 5, 1, 12, 0, 0)
    epoch_str = str(datetime(2023, 5, 1, 12, 0, 0).timestamp())
    assert timestamp_format_to_datetime(epoch_str, "epoch") == datetime(2023, 5, 1, 12, 0, 0)


def test_recursive_replace():
    org_dict = {"key1": "value1", "key2": {"key3": "value3"}}
    substitutions = [("value1", "new_value1"), ("value3", "new_value3")]
    result = recursive_replace(org_dict, substitutions)
    assert result == {"key1": "new_value1", "key2": {"key3": "new_value3"}}


def test_recursive_replace_with_none():
    assert recursive_replace(None, []) is None


def test_get_time_field_from_event_to_dt():
    event = {"timestamp": "2023-05-01T12:00:00Z"}
    config = TimestampFieldConfig(["timestamp"], "%Y-%m-%dT%H:%M:%SZ")
    result = get_time_field_from_event_to_dt(event, config)
    assert isinstance(result, datetime)
    assert result == datetime(2023, 5, 1, 12, 0, 0)


def test_get_time_field_from_event_to_dt_throws_exception():
    event = {"non_timestamp_field": "2023-05-01T12:00:00Z"}
    config = TimestampFieldConfig(["timestamp"], "%Y-%m-%dT%H:%M:%SZ")
    with pytest.raises(DemistoException) as exc_info:
        get_time_field_from_event_to_dt(event, config)
    assert "Timestamp field: ['timestamp'] not found in event" in str(exc_info.value)


def test_is_pagination_needed():
    events = {"next_page": "page2", "has_more": True}
    pagination_logic = PaginationLogic(True, ["next_page"], ["has_more"])
    needed, next_page = is_pagination_needed(events, pagination_logic)
    assert needed is True
    assert next_page == "page2"


def test_is_pagination_not_needed():
    events = {"has_more": False}
    pagination_logic = PaginationLogic(True, ["next_page"], ["has_more"])
    needed, next_page = is_pagination_needed(events, pagination_logic)
    assert needed is False
    assert next_page is None


def test_iso8601_to_datetime_str():
    iso_time = "2023-05-01T12:00:00.1234567Z"
    result = iso8601_to_datetime_str(iso_time)
    assert result == "2023-05-01T12:00:00.123456Z"


def test_parse_json_param():
    json_param_value = '{"key": "value"}'
    result = parse_json_param(json_param_value, "test_param")
    assert result == {"key": "value"}


def test_parse_json_param_none_value():
    result = parse_json_param(None, "test_param")
    assert result is None


def test_organize_events_to_xsiam_format():
    raw_events = {"data": {"events": [{"id": 1, "name": "event1"}, {"id": 2, "name": "event2"}]}}
    events_keys = ["data", "events"]
    expected_output = [{"id": 1, "name": "event1"}, {"id": 2, "name": "event2"}]
    assert organize_events_to_xsiam_format(raw_events, events_keys) == expected_output


def test_organize_events_to_xsiam_format_empty():
    raw_events = {}
    events_keys = ["data", "events"]
    expected_output = []
    assert organize_events_to_xsiam_format(raw_events, events_keys) == expected_output


def test_organize_events_to_xsiam_format_no_events_key():
    raw_events = {"data": {"no_events": [{"id": 1, "name": "event1"}, {"id": 2, "name": "event2"}]}}
    events_keys = ["data", "events"]
    expected_output = []
    assert organize_events_to_xsiam_format(raw_events, events_keys) == expected_output


def test_generate_authentication_headers_basic():
    params = {"authentication": "Basic", "credentials": {"identifier": "user", "password": "pass"}}
    headers = generate_authentication_headers(params)
    assert headers["Authorization"].startswith("Basic ")


@patch("GenericAPIEventCollector.return_error")
@patch("GenericAPIEventCollector.demisto.error")
def test_generate_authentication_headers_basic_no_password(mock_error, mock_return_error):
    params = {"authentication": "Basic", "credentials": {"identifier": "user"}}
    generate_authentication_headers(params)
    mock_error.assert_called_once_with("Password is required for Basic Authentication.")
    mock_return_error.assert_called_once_with("Password is required for Basic Authentication.")


def test_generate_authentication_headers_bearer():
    params = {"authentication": "Bearer", "token": {"password": "test_token"}}
    headers = generate_authentication_headers(params)
    assert headers["Authorization"] == "Bearer test_token"


@patch("GenericAPIEventCollector.return_error")
@patch("GenericAPIEventCollector.demisto.error")
def test_generate_authentication_headers_bearer_no_token(mock_error, mock_return_error):
    params = {"authentication": "Bearer"}
    generate_authentication_headers(params)
    mock_error.assert_called_once_with("API Token is required.")
    mock_return_error.assert_called_once_with("API Token is required.")


def test_generate_authentication_headers_token():
    params = {"authentication": "Token", "token": {"password": "test_token"}}
    headers = generate_authentication_headers(params)
    assert headers["Authorization"] == "Token test_token"


@patch("GenericAPIEventCollector.return_error")
@patch("GenericAPIEventCollector.demisto.error")
def test_generate_authentication_headers_token_no_token(mock_error, mock_return_error):
    params = {"authentication": "Token"}
    generate_authentication_headers(params)
    mock_error.assert_called_once_with("API Token is required.")
    mock_return_error.assert_called_once_with("API Token is required.")


def test_generate_authentication_headers_api_key():
    params = {"authentication": "Api-Key", "token": {"password": "test_token"}}
    headers = generate_authentication_headers(params)
    assert headers["api-key"] == "test_token"


@patch("GenericAPIEventCollector.return_error")
@patch("GenericAPIEventCollector.demisto.error")
def test_generate_authentication_headers_api_key_no_token(mock_error, mock_return_error):
    params = {"authentication": "Api-Key"}
    generate_authentication_headers(params)
    mock_error.assert_called_once_with("API Token is required.")
    mock_return_error.assert_called_once_with("API Token is required.")


def test_generate_authentication_headers_raw_token():
    params = {"authentication": "RawToken", "token": {"password": "test_token"}}
    headers = generate_authentication_headers(params)
    assert headers["Authorization"] == "test_token"


@patch("GenericAPIEventCollector.return_error")
@patch("GenericAPIEventCollector.demisto.error")
def test_generate_authentication_headers_raw_token_no_token(mock_error, mock_return_error):
    params = {"authentication": "RawToken"}
    generate_authentication_headers(params)
    mock_error.assert_called_once_with("API Token is required.")
    mock_return_error.assert_called_once_with("API Token is required.")


def test_generate_authentication_headers_no_auth():
    params = {"authentication": "No Authorization"}
    headers = generate_authentication_headers(params)
    assert headers == {}


@patch("GenericAPIEventCollector.return_error")
@patch("GenericAPIEventCollector.demisto.error")
def test_generate_authentication_headers_invalid_auth(mock_error, mock_return_error):
    params = {"authentication": "InvalidAuth"}
    generate_authentication_headers(params)
    mock_error.assert_called_once_with(
        "Please insert a valid authentication method, options are: Basic, Bearer, Token, Api-Key, RawToken"
        "No Authorization, got: InvalidAuth"
    )
    mock_return_error.assert_called_once_with(
        "Please insert a valid authentication method, options are: Basic, Bearer, Token, Api-Key, RawToken"
        "No Authorization, got: InvalidAuth"
    )


def test_extract_pagination_params():
    params = {"pagination_needed": "true", "pagination_field_name": "next_page", "pagination_flag": "has_more"}
    pagination_logic = extract_pagination_params(params)
    assert pagination_logic.pagination_needed is True
    assert pagination_logic.pagination_field_name == ["next_page"]
    assert pagination_logic.pagination_flag == ["has_more"]


def test_setup_search_events():
    first_fetch_datetime = datetime(2023, 1, 1, 0, 0, 0)
    last_run = {}
    params = {
        "request_data": '{"key": "value"}',
        "request_json": '{"json_key": "json_value"}',
        "query_params": '{"param_key": "param_value"}',
        "pagination_needed": "true",
        "pagination_field_name": "next_page",
        "pagination_flag": "has_more",
        "timestamp_field_name": "timestamp",
        "timestamp_format": "%Y-%m-%dT%H:%M:%SZ",
    }
    timestamp_field_config = TimestampFieldConfig(["timestamp"], "%Y-%m-%dT%H:%M:%SZ")

    last_fetched_datetime, pagination_logic, request_data = setup_search_events(
        first_fetch_datetime, last_run, params, timestamp_field_config
    )

    assert last_fetched_datetime == first_fetch_datetime
    assert pagination_logic == PaginationLogic(True, ["next_page"], ["has_more"])
    assert request_data == RequestData({"key": "value"}, {"json_key": "json_value"}, {"param_key": "param_value"})


def test_setup_search_events_with_last_run():
    first_fetch_datetime = datetime(2023, 1, 1, 0, 0, 0)
    last_run = {"@last_fetched_datetime": "2023-01-02T00:00:00"}
    params = {
        "request_data": '{"key": "value"}',
        "request_json": '{"json_key": "json_value"}',
        "query_params": '{"param_key": "param_value"}',
        "pagination_needed": "true",
        "pagination_field_name": "next_page",
        "pagination_flag": "has_more",
        "timestamp_field_name": "timestamp",
        "timestamp_format": "%Y-%m-%dT%H:%M:%SZ",
    }
    timestamp_field_config = TimestampFieldConfig(["timestamp"], "%Y-%m-%dT%H:%M:%SZ")

    last_fetched_datetime, pagination_logic, request_data = setup_search_events(
        first_fetch_datetime, last_run, params, timestamp_field_config
    )

    assert last_fetched_datetime == datetime(2023, 1, 2, 0, 0, 0)
    assert pagination_logic == PaginationLogic(True, ["next_page"], ["has_more"])
    assert request_data == RequestData({"key": "value"}, {"json_key": "json_value"}, {"param_key": "param_value"})


def test_setup_search_events_first_fetch():
    first_fetch_datetime = datetime(2023, 1, 1, 0, 0, 0)
    last_run = {}
    params = {
        "request_data": '{"key": "value"}',
        "request_json": '{"json_key": "json_value"}',
        "query_params": '{"param_key": "param_value"}',
        "pagination_needed": "true",
        "pagination_field_name": "next_page",
        "pagination_flag": "has_more",
        "timestamp_field_name": "timestamp",
        "timestamp_format": "%Y-%m-%dT%H:%M:%SZ",
        "initial_query_params": '{"initial_param_key": "initial_param_value"}',
        "initial_pagination_params": {
            "pagination_needed": "true",
            "pagination_field_name": "next_page",
            "pagination_flag": "has_more",
        },
        "initial_request_data": '{"initial_key": "initial_value"}',
        "initial_request_json": '{"initial_json_key": "initial_json_value"}',
    }
    timestamp_field_config = TimestampFieldConfig(["timestamp"], "%Y-%m-%dT%H:%M:%SZ")

    last_fetched_datetime, pagination_logic, request_data = setup_search_events(
        first_fetch_datetime, last_run, params, timestamp_field_config
    )

    assert last_fetched_datetime == first_fetch_datetime
    assert pagination_logic == PaginationLogic(True, ["next_page"], ["has_more"])
    assert request_data == RequestData(
        {"initial_key": "initial_value"}, {"initial_json_key": "initial_json_value"}, {"initial_param_key": "initial_param_value"}
    )


@patch("GenericAPIEventCollector.generate_authentication_headers")
def test_generate_headers_basic(mock_generate_authentication_headers):
    params = {
        "authentication": "Basic",
        "credentials": {"identifier": "user", "password": "pass"},
        "add_fields_to_header": '{"Custom-Header": "CustomValue"}',
    }
    mock_generate_authentication_headers.return_value = {"Authorization": "Basic d5Nl4jp3YX2z"}
    headers = generate_headers(params)
    assert headers == {"Authorization": "Basic d5Nl4jp3YX2z", "Custom-Header": "CustomValue"}


@patch("GenericAPIEventCollector.generate_authentication_headers")
def test_generate_headers_bearer(mock_generate_authentication_headers):
    params = {
        "authentication": "Bearer",
        "token": {"password": "test_token"},
        "add_fields_to_header": '{"Custom-Header": "CustomValue"}',
    }
    mock_generate_authentication_headers.return_value = {"Authorization": "Bearer test_token"}
    headers = generate_headers(params)
    assert headers == {"Authorization": "Bearer test_token", "Custom-Header": "CustomValue"}


@patch("GenericAPIEventCollector.generate_authentication_headers")
def test_generate_headers_token(mock_generate_authentication_headers):
    params = {
        "authentication": "Token",
        "token": {"password": "test_token"},
        "add_fields_to_header": '{"Custom-Header": "CustomValue"}',
    }
    mock_generate_authentication_headers.return_value = {"Authorization": "Token test_token"}
    headers = generate_headers(params)
    assert headers == {"Authorization": "Token test_token", "Custom-Header": "CustomValue"}


@patch("GenericAPIEventCollector.generate_authentication_headers")
def test_generate_headers_api_key(mock_generate_authentication_headers):
    params = {
        "authentication": "Api-Key",
        "token": {"password": "test_token"},
        "add_fields_to_header": '{"Custom-Header": "CustomValue"}',
    }
    mock_generate_authentication_headers.return_value = {"api-key": "test_token"}
    headers = generate_headers(params)
    assert headers == {"api-key": "test_token", "Custom-Header": "CustomValue"}


@patch("GenericAPIEventCollector.generate_authentication_headers")
def test_generate_headers_raw_token(mock_generate_authentication_headers):
    params = {
        "authentication": "RawToken",
        "token": {"password": "test_token"},
        "add_fields_to_header": '{"Custom-Header": "CustomValue"}',
    }
    mock_generate_authentication_headers.return_value = {"Authorization": "test_token"}
    headers = generate_headers(params)
    assert headers == {"Authorization": "test_token", "Custom-Header": "CustomValue"}


@patch("GenericAPIEventCollector.generate_authentication_headers")
def test_generate_headers_no_auth(mock_generate_authentication_headers):
    params = {"authentication": "No Authorization", "add_fields_to_header": '{"Custom-Header": "CustomValue"}'}
    mock_generate_authentication_headers.return_value = {}
    headers = generate_headers(params)
    assert headers == {"Custom-Header": "CustomValue"}


@pytest.mark.parametrize(
    "input_str,expected",
    [
        ("12345678901", True),  # 11 chars - should be detected as milliseconds
        ("1234567890", False),  # 10 chars - should not be detected as milliseconds
        ("", False),  # Empty string
    ],
)
def test_is_milliseconds(input_str, expected):
    """Tests the is_milliseconds function with various inputs."""
    result = is_milliseconds(input_str)
    assert result == expected


@pytest.mark.parametrize(
    "input_str,expected",
    [
        ("1234567890123456", True),  # 16 chars - should be detected as microseconds
        ("123456789012345", False),  # 15 chars - should not be detected as microseconds
        ("", False),  # Empty string
        ("12345678901234567", True),  # Longer than 16 chars
    ],
)
def test_is_microseconds(input_str, expected):
    """Tests the is_microseconds function with various inputs."""
    result = is_microseconds(input_str)
    assert result == expected


@patch("GenericAPIEventCollector.demisto.debug")
def test_convert_epoch_to_timestamp_seconds(mock_debug):
    """Test converting epoch in seconds to timestamp."""
    # Unix timestamp for 2022-01-01 00:00:00 UTC
    epoch_seconds = "1640995200"
    expected_datetime = datetime.fromtimestamp(float(epoch_seconds))

    result = convert_epoch_to_timestamp(epoch_seconds)

    assert result == expected_datetime
    # For seconds, debug should not be called
    mock_debug.assert_not_called()


@patch("GenericAPIEventCollector.demisto.debug")
def test_convert_epoch_to_timestamp_milliseconds(mock_debug):
    """Test converting epoch in milliseconds to timestamp."""
    # '1640995200000' is 2022-01-01 00:00:00 UTC in milliseconds
    epoch_milliseconds = "1640995200000"
    expected_datetime = datetime.fromtimestamp(float(epoch_milliseconds) / 1000)

    result = convert_epoch_to_timestamp(epoch_milliseconds)

    assert result == expected_datetime
    # For milliseconds, debug should be called with the appropriate message
    mock_debug.assert_called_once_with(f"converting {epoch_milliseconds} epoch milliseconds to timestamp")


@patch("GenericAPIEventCollector.demisto.debug")
def test_convert_epoch_to_timestamp_microseconds(mock_debug):
    """Test converting epoch in microseconds to timestamp."""
    # '1640995200000000' is 2022-01-01 00:00:00 UTC in microseconds
    epoch_microseconds = "1640995200000000"
    expected_datetime = datetime.fromtimestamp(float(epoch_microseconds) / 1_000_000)

    result = convert_epoch_to_timestamp(epoch_microseconds)

    assert result == expected_datetime
    # For microseconds, debug should be called with the appropriate message
    mock_debug.assert_called_once_with(f"converting {epoch_microseconds} epoch microseconds to timestamp")


@patch("GenericAPIEventCollector.demisto.debug")
def test_convert_epoch_to_timestamp_boundary_conditions(mock_debug):
    """Test boundary conditions for epoch conversion."""
    # Test with exactly 10 chars (milliseconds' threshold)
    epoch_10_chars = "12345678901"  # 10 digits - should be treated as milliseconds
    result_10 = convert_epoch_to_timestamp(epoch_10_chars)
    expected_10 = datetime.fromtimestamp(float(epoch_10_chars) / 1_000)

    assert result_10 == expected_10
    # Reset the mock to check the next call
    mock_debug.reset_mock()

    # Test with exactly 16 chars (microseconds' threshold)
    epoch_16_chars = "1234567890123456"  # 16 digits - should be treated as microseconds
    result_16 = convert_epoch_to_timestamp(epoch_16_chars)
    expected_16 = datetime.fromtimestamp(float(epoch_16_chars) / 1_000_000)

    assert result_16 == expected_16