Elasticsearch v2

Search for and analyze data in real time. Supports version 6 and later.

Database · Elasticsearch

Details

IDElasticsearch v2
ProviderElastic
CategoryDatabase
From Version5.0.0
Docker Imagedemisto/elasticsearch:1.0.0.11195141
Supported ModulesAgentix Cloud Runtime Security XSIAM EDR Cortex Cloud Attack Surface Management Cloud Posture Security Exposure Management

README

Elasticsearch_v2 integration is used to search for and analyze data in real time.
Supports version 6 and later.
This integration was integrated and tested with versions 6.6.2, 7.3, 8.4.1 of Elasticsearch.

Permissions

The permissions required to use this integration depends on which operations you need to perform. The API user should have the same permissions a regular user would have in order to access the data via the UI. Following are the permissions needed for certain commands:

  • !es-eql-search/search/es-search/fetch-incidents - If the Elasticsearch security features are enabled, you must have the read index privilege for the target data stream, index, or alias.
  • !get-mapping-fields - If the Elasticsearch security features are enabled, you must have the view_index_metadata or manage index privilege for the target data stream, index, or alias.
  • !es-index - If the Elasticsearch security features are enabled, you must have the write index privilege for the target data stream, index, or alias.
  • !es-kibana-* commands - To execute Kibana commands, you must have the necessary privileges for the applicable resource. The privilege required depends on the command you are using. Kibana API endpoints are gated by feature privileges granted at one of two levels: Read (GET / list / view operations) and All (POST / PUT / PATCH / DELETE operations). For rules and cases, the privilege is scoped to the feature that owns the object — a Security-owned object requires the Security feature privilege, an Observability-owned object requires the Observability feature privilege, and a stack-owned object requires the Management / Stack Rules feature privilege.

Configure Elasticsearch v2 in Cortex

Parameter Description Required
Server URL The Elasticsearch server to which the integration connects. Ensure that the URL includes the correct Elasticsearch port. The default port for Elasticsearch v7 and below is 9200. Use the Server URL for on-premises deployments. False
Kibana Server URL The Kibana server used by the es-kibana-* commands and by fetching and mirroring security alerts and cases. Required for on-premises (self-managed) deployments, where Kibana is usually hosted separately from Elasticsearch (for example, https://kibana.example.com:5601). If left empty, the Kibana URL is derived from the Server URL, which is supported only for Elastic Cloud deployments whose hostname contains “.es.” (the segment is replaced with “.kb.”). False
Authorization type Select the authentication type and enter the appropriate credentials:
- Basic Auth: Enter Username and Password.
- Bearer Auth: Enter Username and Password.
- API Key Auth: Enter the API Key ID and API Key.		 True
API key ID Use for API key auth False
API Key Use for API key auth False
Username Use for API Key auth. Optionally you can use Username as an API key ID and Password as an API key for Basic auth. False
Password Use for API Key auth. Optionally you can use Username as an API key ID and Password as an API key for Basic auth. False
Trust any certificate (not secure)   False
Use system proxy settings   False
Client type In some hosted ElasticSearch environments, the standard ElasticSearch client is not supported. If you encounter any related client issues, please consider using the OpenSearch client type. False
Index to fetch incidents from CSV False
Query String Query string uses the Lucene syntax.
For more information about the Lucene syntax see here.
The query will be used when fetching incidents. Index time field will be used as a filter in the query.

The integration test button doesn’t fully test the fetch incidents validity. To verify that the instance is set up correctly for fetching incidents, run the ‘!es-integration-health-check’ command.		 False
Fields to fetch (_source already included) A comma-separated list of Elasticsearch fields to retrieve in addition to the default _source fields. For example: host.hostname, host.id. False
Index time field The time field used for sorting and limiting results. If using a nested field, separate field names with dot notation. False
Raw Query Raw Query allows raw DSL queries and will override the ‘Query String’ Lucene syntax string.
Results will not be filtered or sorted. Use additional parameters in the raw query for these purposes.
For more information about Query DSL see here.

The integration test button doesn’t fully test the fetch incidents validity. To verify that the instance is set up correctly for fetching incidents, run the ‘!es-integration-health-check’ command.		 False
Time field type 3 formats supported:
Simple-Date - A plain date string. You must specify the format in which the date is stored.
For more information about time formatting, see here.
 Timestamp-Second - A numeric value representing the number of seconds since the Unix epoch (00:00:00 UTC on 1 January 1970). Example: ‘1572164838’
* Timestamp-Milliseconds - A numeric value representing the number of milliseconds since the Unix epoch. Example: ‘1572164838123’		 False
Map JSON fields into labels   False
First fetch timestamp <number> <time unit>, e.g., 12 hours, 7 days False
The maximum number of results per fetch   False
Request timeout (in seconds).   False
Incident type   False
Fetch incidents   False
Space ID The default Kibana space ID to use for es-kibana-* commands. Used to derive the Kibana base URL from the Server URL. If a space_id argument is provided to a command, it overrides this value. See Spaces for more information. False
Fetch incident types The type of incident to fetch from Elasticsearch (Elasticsearch Entity, Elasticsearch Security Alert, or Elasticsearch Case). False
Incident Mirroring Direction Mirroring works only for the Elasticsearch Security Alert and Elasticsearch Case incident types. False
Fetch cases by Severity Case severities to fetch. If none are chosen, all severity levels will be returned. Relevant only for incident type Elasticsearch Case. False
Fetch cases by Status Case statuses to fetch. If none are chosen, in-progress and open will be returned. Relevant only for incident type Elasticsearch Case. False
Fetch alerts for case When enabled, fetches alerts associated with each Elasticsearch Case. You must have read privileges for the Cases feature in the Management, Observability, or Security section of the Kibana feature privileges. False
Close Mirrored XSOAR Incident Whether closing the Elasticsearch alert or case is mirrored in Cortex XSOAR. False
Close Mirrored Elasticsearch Incident Whether closing the Cortex XSOAR incident is mirrored in Elasticsearch. False

Fetch

Use the Fetch incident types parameter to select what to fetch from Elasticsearch: the default Elasticsearch Entity, Elasticsearch Security Alert, or Elasticsearch Case.

Mirroring is only available for Elasticsearch Security Alerts and Cases.

Fetch

The default fetch behavior. Configure the Index to fetch incidents from, Query String, and/or Raw Query fields to fetch generic Elasticsearch documents as incidents.

Fetch Security Alerts

To fetch security alerts, use the Raw Query field (DSL query). The index must follow one of these patterns: .internal.alerts-security.alerts-* or .siem-signals-*.

Fetch Cases

Use the Fetch cases by Severity parameter to filter cases by the required severity and the Fetch cases by Status parameter to filter cases by the required status. Enable Fetch alerts for case to also retrieve the alerts associated with each case.

Incident Mirroring

You can enable incident mirroring between Cortex XSOAR incidents and Elasticsearch (Kibana) Security Alerts and Cases.

Mirroring is only available for Elasticsearch Security Alert and Elasticsearch Case incident types.

To set up mirroring:

  1. Enable Fetch incidents in your instance configuration.
  2. In the Fetch incident types parameter, select Elasticsearch Security Alert or Elasticsearch Case.
  3. In the Incident Mirroring Direction parameter, select in which direction the incidents should be mirrored:
    • Incoming - Changes in Elasticsearch (status, tags, workflow reason, case fields) are reflected in Cortex XSOAR.
    • Outgoing - Changes in Cortex XSOAR are reflected in Elasticsearch.
    • Incoming And Outgoing - Changes are mirrored in both directions.
  4. Optionally, select Close Mirrored XSOAR Incident to close the Cortex XSOAR incident when the Elasticsearch alert/case is closed, and Close Mirrored Elasticsearch Incident to close the Elasticsearch alert/case when the Cortex XSOAR incident is closed.

Notes:

  • The mirroring is affected by the following integration parameters: Incident Mirroring Direction, Close Mirrored XSOAR Incident, and Close Mirrored Elasticsearch Incident.
  • To ensure mirroring works as expected, mappers are required so that the fields are properly mapped to and from Elasticsearch. Select the appropriate incoming and outgoing mappers per incident type (Elasticsearch Security Alert - Incoming/Outgoing Mapper and Elasticsearch Case - Incoming/Outgoing Mapper).
  • Set the Fetch incident types parameter even when mirroring only. It is used to resolve the incident type when the remote lookup is unavailable; without it, changes may not be mirrored out.
  • Mirroring out always uses the Kibana API, because the security alert indices are Kibana system indices and writing to them directly would skip Kibana’s bookkeeping (such as kibana.alert.workflow_status_updated_at, which mirroring in relies on). The configured credentials therefore need the Security → Alerts: All Kibana privilege, and the Space ID must match the space that owns the alerts. If either is wrong, the fetch keeps working while mirroring out reports that 0 alerts were updated.

Which Security Alert fields can be mirrored

Elasticsearch only allows a detection alert’s workflow status, status reason, and workflow tags to be changed. These are the only fields mirrored out:

Cortex XSOAR incident field Elasticsearch field
Elasticsearch Workflow Alert Status kibana.alert.workflow_status
Elasticsearch Workflow Alert Status Reason kibana.alert.workflow_reason
Tags kibana.alert.workflow_tags

Other fields, such as severity, risk score, or the detection rule, cannot be pushed back to Elasticsearch because no API exists to change them on an existing alert. Editing those fields in Cortex XSOAR updates the local incident only, and an error is written to the log explaining that nothing was mirrored out. These fields are still mirrored in (from Elasticsearch to Cortex XSOAR).

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.

es-search

Queries an index.

Base Command

es-search

Input

Argument Name Description Required
index The index in which to perform a search. Required
query The string to query (in Lucene syntax). Possible values are: . Optional
fields A comma-separated list of document fields to fetch. If empty, the entire document is fetched. Optional
explain Calculates an explanation of a score for a query. For example, “value:1.6943597”. Possible values are: true, false. Default is false. Optional
page The page number from which to start a search. Default is 0. Optional
size The number of documents displayed per page. Can be an integer between “1” and “10,000”. Default is 100. Optional
sort-field The field by which to sort the results table. The supported result types are boolean, numeric, date, and keyword fields. Keyword fields require the doc_values parameter to be set to “true” from the Elasticsearch server. Possible values are: . Optional
sort-order The order by which to sort the results table. The results tables can only be sorted if a sort-field is defined. Possible values are: asc, desc. Default is asc. Optional
query_dsl Will overwrite the ‘query’ arguments. Optional
timestamp_range_start The starting time of the time range. Optional
timestamp_range_end The ending time of the time range. Optional
timestamp_field Timestamp field name. Default is @timestamp. Optional

Context Output

Path Type Description
Elasticsearch.Search.Results._index String The index to which the document belongs.
Elasticsearch.Search.Results._id String The ID of the document.
Elasticsearch.Search.Results._type String The mapping type of the document.
Elasticsearch.Search.max_score Number The maximum relevance score of a query.
Elasticsearch.Search.Query String The query performed in the search.
Elasticsearch.Search.total.value Number The number of search results.
Elasticsearch.Search.Results._score Number The relevance score of the search result.
Elasticsearch.Search.Index String The index in which the search was performed.
Elasticsearch.Search.Server String The server on which the search was performed.
Elasticsearch.Search.timed_out Boolean Whether the search stopped due to a timeout.
Elasticsearch.Search.took Number The time in milliseconds taken for the search to complete.
Elasticsearch.Search.Page Number The page number from which the search started.
Elasticsearch.Search.Size Number The maximum number of scores that a search can return.

search

Searches an index.

Base Command

search

Input

Argument Name Description Required
index The index in which to perform a search. Required
query The string to query (in Lucene syntax). Possible values are: . Optional
fields A comma-separated list of document fields to fetch. If empty, fetches the entire document. Optional
explain Calculates an explanation of a score for a query. For example, “value:1.6943597”. Possible values are: true, false. Default is false. Optional
page The page number from which to start a search. Default is 0. Optional
size The number of documents displayed per page. Can be an integer between “1” and “10,000”. Default is 100. Optional
sort-field The field by which to sort the results table. The supported result types are boolean, numeric, date, and keyword fields. Keyword fields require the doc_values parameter to be set to “true” from the Elasticsearch server. Possible values are: . Optional
sort-order The order by which to sort the results table. The results tables can only be sorted if a sort-field is defined. Possible values are: asc, desc. Default is asc. Optional
timestamp_field Timestamp field name. Default is @timestamp. Optional

Context Output

Path Type Description
Elasticsearch.Search.Results._index String The index to which the document belongs.
Elasticsearch.Search.Results._id String The ID of the document.
Elasticsearch.Search.Results._type String The mapping type of the document.
Elasticsearch.Search.max_score Number The maximum relevance score of a query.
Elasticsearch.Search.Query String The query performed in the search.
Elasticsearch.Search.total.value Number The number of search results.
Elasticsearch.Search.Results._score Number The relevance score of the search result.
Elasticsearch.Search.Index String The index in which the search was performed.
Elasticsearch.Search.Server String The server on which the search was performed.
Elasticsearch.Search.timed_out Boolean Whether the search stopped due to a time out.
Elasticsearch.Search.took Number The time in milliseconds taken for the search to complete.
Elasticsearch.Search.Page Number The page number from which the search started.
Elasticsearch.Search.Size Number The maximum number of scores that a search can return.

get-mapping-fields

Returns the schema of the index to fetch from. This commmand should be used for debugging purposes.

Base Command

get-mapping-fields

Input

There are no input arguments for this command.

Context Output

There is no context output for this command.

es-eql-search

Search using EQL query

Base Command

es-eql-search

Input

Argument Name Description Required
index The index in which to perform a search. Required
query The string to query (in Lucene syntax). Required
fields A comma-separated list of document fields to fetch. If empty, fetches the entire document. Optional
sort-tiebreaker If two or more events share the same timestamp, Elasticsearch uses a tiebreaker field value to sort the events in ascending order. Optional
filter Filter using query DSL. Optional
event_category_field The event category field. Default is event.category. Optional
size The number of documents displayed per page. Can be an integer between “1” and “10,000”. Default is 100. Optional
timestamp_range_start The starting time of the time range. Optional
timestamp_range_end The ending time of the time range. Optional
timestamp_field Timestamp field name. Default is @timestamp. Optional

Context Output

Path Type Description
Elasticsearch.Search.Results._index String The index to which the document belongs.
Elasticsearch.Search.Results._id String The ID of the document.
Elasticsearch.Search.Results._type String The mapping type of the document.
Elasticsearch.Search.max_score Number The maximum relevance score of a query.
Elasticsearch.Search.Query String The query performed in the search.
Elasticsearch.Search.total.value Number The number of search results.
Elasticsearch.Search.Results._score Number The relevance score of the search result.
Elasticsearch.Search.Index String The index in which the search was performed.
Elasticsearch.Search.Server String The server on which the search was performed.
Elasticsearch.Search.timed_out Boolean Whether the search stopped due to a timeout.
Elasticsearch.Search.took Number The time in milliseconds taken for the search to complete.
Elasticsearch.Search.Page Number The page number from which the search started.
Elasticsearch.Search.Size Number The maximum number of scores that a search can return.

es-index

Indexes a document into an Elasticsearch index.

Base Command

es-index

Input

Argument Name Description Required
index_name The name of the index to ingest into. Required
document The document object (JSON format) to be indexed. See Elasticsearch documentation for further information about indexing documents via the Elasticsearch client. Required
id The ID of the indexed document (will be generated if empty). The document will be updated if one with a corresponding ID exists. Optional

Context Output

Path Type Description
Elasticsearch.Index.id String The ID of the indexed document.
Elasticsearch.Index.index String The name of the index which the document was ingested to.
Elasticsearch.Index.version Number The version number of the indexed document.
Elasticsearch.Index.result String The result of the index operation.

Command Example

!es-index index_name=test-xsoar document="{\"name\":\"test\"}" id=1234

Context Example

{
    "Elasticsearch": {
        "Index": {
            "id": "1234",
            "index": "test-xsoar",
            "version": 1,
            "result": "created"
        }
    }
}

Human Readable Output

Indexed document

ID Index name Version Result
1234 test-xsoar 1 created

es-integration-health-check

Returns the health status of the integration. This commmand should be used for debugging purposes.

Base Command

es-integration-health-check

Input

There are no input arguments for this command.

Context Output

There is no context output for this command.

es-get-indices-statistics

Returns Elasticsearch indices statistics and information. This command is not supported for client type OpenSearch.

Base Command

es-get-indices-statistics

Input

Argument Name Description Required
limit The maximum number of indices to return. Default is 50. Optional
all_results Whether to retrieve all the Elasticsearch indices. If true, the “limit” argument will be ignored. Optional

Context Output

Path Type Description
Elasticsearch.IndexStatistics.Name String The name of the index.
Elasticsearch.IndexStatistics.Status String The status of the index.
Elasticsearch.IndexStatistics.Health String The health status of the index.
Elasticsearch.IndexStatistics.UUID String The UUID of the index.
Elasticsearch.IndexStatistics.DocumentsCount Number The number of documents that are indexed in the index.
Elasticsearch.IndexStatistics.DocumentsDeleted Number The number of documents that were deleted from the index.

es-esql-search

Search using ES|QL query (Elasticsearch 8.11 and above).

Base Command

es-esql-search

Input

Argument Name Description Required  
query The ES|QL query string to execute using piped syntax (for example, FROM index WHERE field == “value”). Required
limit Maximum number of results to return. Optional  

Context Output

Path Description Type
Elasticsearch.ESQLSearch ES|QL search result unknown

Command Example

!es-esql-search query="FROM logs-* | WHERE host.name == \"web-01\" | LIMIT 5"

Context Example

{
    "Elasticsearch": {
        "ESQLSearch": [
            {
                "@timestamp": "2024-01-15T10:23:45.000Z",
                "host.name": "web-01",
                "message": "Connection established"
            }
        ]
    }
}

Human Readable Output

Search query

Query Total
FROM logs-* | WHERE host.name == “web-01” | LIMIT 5 1

Results

@timestamp host.name message
2024-01-15T10:23:45.000Z web-01 Connection established

es-kibana-alerting-health-get

Retrieves the health of the Kibana alerting framework.

Base Command

es-kibana-alerting-health-get

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.AlertingHealth.is_sufficiently_secure Boolean Whether the alerting framework is sufficiently secure.
Elasticsearch.Kibana.AlertingHealth.has_permanent_encryption_key Boolean Whether a permanent encryption key is configured.
Elasticsearch.Kibana.AlertingHealth.alerting_framework_health.decryption_health.status String The decryption health status.
Elasticsearch.Kibana.AlertingHealth.alerting_framework_health.execution_health.status String The execution health status.
Elasticsearch.Kibana.AlertingHealth.alerting_framework_health.read_health.status String The read health status.

Command Example


#### Context Example

```json
{
    "Elasticsearch": {
        "Kibana": {
            "AlertingHealth": {
                "is_sufficiently_secure": true,
                "has_permanent_encryption_key": true,
                "alerting_framework_health": {
                    "decryption_health": {"status": "ok"},
                    "execution_health": {"status": "ok"},
                    "read_health": {"status": "ok"}
                }
            }
        }
    }
}

Human Readable Output

Kibana Alerting Framework Health

Is sufficiently secure Has permanent encryption key Decryption status Execution status Read status
true true ok ok ok

es-kibana-rule-types-list

Retrieves all rule types available in Kibana.

Base Command

es-kibana-rule-types-list

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.RuleType.id String The rule type ID.
Elasticsearch.Kibana.RuleType.name String The rule type name.
Elasticsearch.Kibana.RuleType.category String The rule type category.
Elasticsearch.Kibana.RuleType.producer String The rule type producer.
Elasticsearch.Kibana.RuleType.action_groups.id String The action group ID.

Command Example


#### Context Example

```json
{
    "Elasticsearch": {
        "Kibana": {
            "RuleType": [
                {
                    "id": ".index-threshold",
                    "name": "Index threshold",
                    "category": "management",
                    "producer": "stackAlerts",
                    "action_groups": [{"id": "threshold met"}, {"id": "recovered"}]
                }
            ]
        }
    }
}

Human Readable Output

Kibana Rule Types

Rule type ID Name Category Producer Action Group Id
.index-threshold Index threshold management stackAlerts threshold met, recovered

es-kibana-rule-list

Retrieves information about rules.

Base Command

es-kibana-rule-list

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
rule_id The identifier for the rule. Optional
search An Elasticsearch simple_query_string query that filters the objects in the response. Optional
default_search_operator The default operator to use for the simple_query_string. Optional
search_fields The fields to perform the simple_query_string parsed query against. Optional
sort_field Determines which field is used to sort the results. Optional
sort_order Determines the sort order. Optional
has_reference_id Filters the rules that have a relation with the reference objects with a specific identifier. Optional
has_reference_type Filters the rules that have a relation with the reference objects with a specific type. Optional
fields The fields to return in the attributes key of the response. Optional
filter A KQL string that you filter with an attribute from your saved object. Optional
filter_consumers List of consumers to filter. Optional
page The page number from which to start a search. Optional
size The number of rules to return per page. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.Rule.id String The rule ID.
Elasticsearch.Kibana.Rule.enabled Boolean Whether the rule is enabled.
Elasticsearch.Kibana.Rule.name String The rule name.
Elasticsearch.Kibana.Rule.rule_type_id String The rule type ID.
Elasticsearch.Kibana.Rule.created_at Date The creation date of the rule.

Command Example

!es-kibana-rule-list rule_id="1234"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "Rule": {
                "id": "1234",
                "enabled": true,
                "name": "CPU threshold alert",
                "rule_type_id": ".index-threshold",
                "created_at": "2024-01-10T08:00:00.000Z"
            }
        }
    }
}

Human Readable Output

Kibana Rules

Rule ID Enabled Name Type ID Creation date
1234 true CPU threshold alert .index-threshold 2024-01-10T08:00:00.000Z

es-kibana-rule-enable

Enable a rule.

Base Command

es-kibana-rule-enable

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
rule_id The identifier for the rule. Required

Context Output

There is no context output for this command.

Command Example

!es-kibana-rule-enable rule_id="1234"

Context Example

{}

Human Readable Output

The rule 1234 has been successfully enabled.

es-kibana-rule-disable

Disable a rule.

Base Command

es-kibana-rule-disable

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
rule_id The identifier for the rule. Required

Context Output

There is no context output for this command.

Command Example

!es-kibana-rule-disable rule_id="1234"

Context Example

{}

Human Readable Output

The rule 1234 has been successfully disabled.

es-kibana-rule-update

Update a rule.

Base Command

es-kibana-rule-update

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
rule_id The identifier for the rule. Required
alert_delay_active The number of consecutive runs that must meet the rule conditions. Optional
artifacts_dashboards_id Not more than 10 elements. Optional
artifacts_investigation_guide_blob Maximum length is 10000. Optional
consumer The name of the application or feature that owns the rule. Optional
enabled Indicates whether you want the rule to run on an interval basis after it is created. Optional
flapping_enabled Determines whether the rule can enter the flapping state. Optional
flapping_look_back_window The minimum number of runs in which the threshold must be met. Minimum value is 2, maximum value is 20. Optional
flapping_status_change_threshold The minimum number of times an alert must switch states within the defined look back window time. Optional
name The name of the rule. Optional
notify_when Indicates how frequently rule actions are triggered. Optional
schedule_interval The interval is specified in seconds, minutes, hours, or days. Optional
tags The tags for the rule. Optional
entry_id Entry ID for the file containing the request JSON. If provided, other parameters won’t be considered. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.Rule.id String The rule ID.

Command Example

!es-kibana-rule-update rule_id="1234" name="Updated CPU alert" schedule_interval="5m"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "Rule": {
                "id": "1234",
                "name": "Updated CPU alert",
                "enabled": true,
                "rule_type_id": ".index-threshold",
                "created_at": "2024-01-10T08:00:00.000Z"
            }
        }
    }
}

Human Readable Output

The rule 1234 has been successfully changed.

Rule ID Changed fields
1234 name, schedule

es-kibana-rule-alert-mute

Mute an alert, or mute all alerts for a rule.

Base Command

es-kibana-rule-alert-mute

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
rule_id The identifier for the rule. Required
alert_id The identifier for the alert. Optional
validate_alerts_existence Whether to validate the existence of the alert. Optional
mute_all Whether to mute all alerts. Optional

Context Output

There is no context output for this command.

Command Example

!es-kibana-rule-alert-mute rule_id="1234" mute_all="true"

Context Example

{}

Human Readable Output

The alerts have been successfully muted.

es-kibana-rule-alert-unmute

Unmute an alert, or unmute all alerts for a rule.

Base Command

es-kibana-rule-alert-unmute

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
rule_id The identifier for the rule. Required
alert_id The identifier for the alert. Optional
unmute_all Whether to unmute all alerts. Optional

Context Output

There is no context output for this command.

Command Example

!es-kibana-rule-alert-unmute rule_id="1234" unmute_all="true"

Context Example

{}

Human Readable Output

The alerts have been successfully unmuted.

es-kibana-detection-alert-status-set

Set the status of one or more detection alerts.

Base Command

es-kibana-detection-alert-status-set

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
status The new status of the detection alert(s). Required
signal_ids List of alert IDs. Use field _id on alert document or kibana.alert.uuid. Optional
query An Elasticsearch query used to select which alerts to update, as an alternative to signal_ids. Optional
reason The reason for the status change. Optional
conflicts Determines how version conflicts should be handled. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.SecurityAlertSetStatus.total Number The total number of alerts matched by the request.
Elasticsearch.Kibana.SecurityAlertSetStatus.updated Number The number of alerts that were updated.

Command Example

!es-kibana-detection-alert-status-set status="closed" signal_ids="1234,5678"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "SecurityAlertSetStatus": {
                "total": 2,
                "updated": 2
            }
        }
    }
}

Human Readable Output

Kibana Detection Alert Status Update

Total Updated
2 2

es-kibana-case-create

Creates a new case in Kibana.

Base Command

es-kibana-case-create

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
assignee_uid An array containing users that are assigned to the case. Not more than 10 elements. Optional
category A word or phrase that categorizes the case. Maximum length is 50. Optional
connector_fields A JSON object containing the connector fields. To create a case without a connector, specify null. Optional
connector_id The identifier for the connector. To create a case without a connector, use none. Optional
connector_name The name of the connector. To create a case without a connector, use none. Optional
connector_type The type of connector. To create a case without a connector, use none. Optional
connector_fields_issue_type_jira The type of issue. Optional
connector_fields_parent_jira The key of the parent issue, when the issue type is sub-task. Optional
connector_fields_priority_jira The priority of the issue. Optional
connector_fields_issue_types_resilient The type of incident. Optional
connector_fields_severity_code_resilient The severity code of the incident. Optional
connector_fields_category_servicenow The category of the incident. Optional
connector_fields_impact_servicenow The effect an incident had on business. Optional
connector_fields_severity_servicenow The severity of the incident. Optional
connector_fields_subcategory_servicenow The subcategory of the incident. Optional
connector_fields_urgency_servicenow The extent to which the incident resolution can be delayed. Optional
connector_fields_dest_ip_servicenow Indicates whether cases will send a comma-separated list of destination IPs. Optional
connector_fields_malware_hash_servicenow Indicates whether cases will send a comma-separated list of malware hashes. Optional
connector_fields_malware_url_servicenow Indicates whether cases will send a comma-separated list of malware URLs. Optional
connector_fields_priority_servicenow The priority of the issue. Optional
connector_fields_source_ip_servicenow Indicates whether cases will send a comma-separated list of source IPs. Optional
connector_fields_case_id_swimlane The case identifier for Swimlane connectors. Optional
custom_key The unique identifier for the custom field. The key value must exist in the case configuration settings. Optional
custom_type The custom field type. It must match the type specified in the case configuration settings. Optional
custom_value The custom field value (string or boolean). Optional
description The description for the case. Maximum length is 30000. Optional
owner The application that owns the cases: Stack Management, Observability, or Elastic Security. Required
extract_observables When true, observables (e.g. IPs, hashes, URLs) are automatically extracted from case comments. Optional
sync_alerts Turns alert syncing on or off. Optional
severity The severity of the case. Optional
tags The words and phrases that help categorize cases. Not more than 200 elements. Optional
title A title for the case. Maximum length is 160. Optional
entry_id Entry ID for the file containing the full request JSON. If provided, other parameters won’t be considered. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.Case.title String The title of the case.
Elasticsearch.Kibana.Case.id String The case ID.
Elasticsearch.Kibana.Case.description String The description of the case.
Elasticsearch.Kibana.Case.owner String The owner of the case.
Elasticsearch.Kibana.Case.severity String The severity of the case.
Elasticsearch.Kibana.Case.status String The status of the case.
Elasticsearch.Kibana.Case.created_at Date The creation date of the case.
Elasticsearch.Kibana.Case.connector.type String The connector type of the case.

Command Example

!es-kibana-case-create owner="securitySolution" title="Suspicious login" description="Multiple failed logins detected" severity="medium"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "Case": {
                "id": "1234",
                "title": "Suspicious login",
                "description": "Multiple failed logins detected",
                "owner": "securitySolution",
                "severity": "medium",
                "status": "open",
                "created_at": "2024-01-15T10:00:00.000Z",
                "connector": {"type": ".none"}
            }
        }
    }
}

Human Readable Output

Kibana Case

Title ID Description Owner Severity Status Created at Connector type
Suspicious login 1234 Multiple failed logins detected securitySolution medium open 2024-01-15T10:00:00.000Z .none

es-kibana-case-update

Update cases in Kibana.

Base Command

es-kibana-case-update

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
case_id The identifier for the case. Required
version The current version of the case. To determine this value, use es-kibana-case-list. Required
assignee_uid An array containing users that are assigned to the case. Not more than 10 elements. Optional
category A word or phrase that categorizes the case. Maximum length is 50. Optional
close_reason The close reason to sync to attached alerts when closing the case. Optional
connector_fields A JSON object containing the connector fields. To remove a connector, specify null. Optional
connector_id The identifier for the connector. To remove a connector, specify none. Optional
connector_name The name of the connector. To remove a connector, specify none. Optional
connector_type The type of connector. To remove a connector, specify none. Optional
connector_fields_issue_type_jira The type of issue. Optional
connector_fields_parent_jira The key of the parent issue, when the issue type is sub-task. Optional
connector_fields_priority_jira The priority of the issue. Optional
connector_fields_issue_types_resilient The type of incident. Optional
connector_fields_severity_code_resilient The severity code of the incident. Optional
connector_fields_category_servicenow The category of the incident. Optional
connector_fields_impact_servicenow The effect an incident had on business. Optional
connector_fields_severity_servicenow The severity of the incident. Optional
connector_fields_subcategory_servicenow The subcategory of the incident. Optional
connector_fields_urgency_servicenow The extent to which the incident resolution can be delayed. Optional
connector_fields_dest_ip_servicenow Indicates whether cases will send a comma-separated list of destination IPs. Optional
connector_fields_malware_hash_servicenow Indicates whether cases will send a comma-separated list of malware hashes. Optional
connector_fields_malware_url_servicenow Indicates whether cases will send a comma-separated list of malware URLs. Optional
connector_fields_priority_servicenow The priority of the issue. Optional
connector_fields_source_ip_servicenow Indicates whether cases will send a comma-separated list of source IPs. Optional
connector_fields_case_id_swimlane The case identifier for Swimlane connectors. Optional
custom_key The unique identifier for the custom field. The key value must exist in the case configuration settings. Optional
custom_type The custom field type. It must match the type specified in the case configuration settings. Optional
custom_value The custom field value (string or boolean). Optional
description The description for the case. Maximum length is 30000. Optional
extract_observables When true, observables (e.g. IPs, hashes, URLs) are automatically extracted from case comments. Optional
sync_alerts Turns alert syncing on or off. Optional
severity The severity of the case. Optional
status The status of the case. Optional
tags The words and phrases that help categorize cases. Not more than 200 elements. Optional
title A title for the case. Maximum length is 160. Optional
entry_id Entry ID for the file containing the request JSON. If provided, other parameters won’t be considered. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.Case.title String The title of the case.
Elasticsearch.Kibana.Case.id String The case ID.
Elasticsearch.Kibana.Case.description String The description of the case.
Elasticsearch.Kibana.Case.owner String The owner of the case.
Elasticsearch.Kibana.Case.severity String The severity of the case.
Elasticsearch.Kibana.Case.status String The status of the case.
Elasticsearch.Kibana.Case.created_at Date The creation date of the case.
Elasticsearch.Kibana.Case.connector.type String The connector type of the case.

Command Example

!es-kibana-case-update case_id="1234" version="WzEsMV0=" title="Suspicious login - updated" severity="high"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "Case": {
                "id": "1234",
                "title": "Suspicious login - updated",
                "description": "Multiple failed logins detected",
                "owner": "securitySolution",
                "severity": "high",
                "status": "open",
                "created_at": "2024-01-15T10:00:00.000Z",
                "connector": {"type": ".none"}
            }
        }
    }
}

Human Readable Output

Kibana Case(s) Updated

Title ID Description Owner Severity Status Created at Connector type
Suspicious login - updated 1234 Multiple failed logins detected securitySolution high open 2024-01-15T10:00:00.000Z .none

es-kibana-case-delete

Deletes one or more cases by ID.

Base Command

es-kibana-case-delete

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
case_id The cases that you want to remove. To get the case identifiers, use es-kibana-case-list. Required

Context Output

There is no context output for this command.

Command Example

!es-kibana-case-delete case_id="1234"

Context Example

{}

Human Readable Output

The cases 1234 have been successfully deleted.

es-kibana-case-list

Retrieves the details of Kibana cases.

Base Command

es-kibana-case-list

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
case_id The identifier for the case. To retrieve case IDs, use the search cases API (leave empty). Optional
assignees Filters the returned cases by assignees. Valid values are none or unique identifiers for the user profiles. Optional
category Filters the returned cases by category. Optional
default_search_operator The default operator to use for the simple_query_string. Optional
search An Elasticsearch simple_query_string query that filters the objects in the response. Optional
from Returns only cases that were created after a specific date (KQL data range or date match expression). Optional
to Returns only cases that were created before a specific date (KQL data range or date match expression). Optional
owner A filter to limit the response to a specific set of applications. Optional
reporters Filters the returned cases by the user name of the reporter. Optional
search_fields The fields to perform the simple_query_string parsed query against. Optional
severity The severity of the case. Optional
sort_field Determines which field is used to sort the results. Optional
sort_order Determines the sort order. Optional
status Filters the returned cases by state. Optional
tags Filters the returned cases by tags. Optional
page The page number from which to start a search. Optional
size The number of items to return. Limited to 100 items. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.Case.title String The title of the case.
Elasticsearch.Kibana.Case.id String The case ID.
Elasticsearch.Kibana.Case.description String The description of the case.
Elasticsearch.Kibana.Case.owner String The owner of the case.
Elasticsearch.Kibana.Case.severity String The severity of the case.
Elasticsearch.Kibana.Case.status String The status of the case.
Elasticsearch.Kibana.Case.created_at Date The creation date of the case.
Elasticsearch.Kibana.Case.connector.type String The connector type of the case.

Command Example


#### Context Example

```json
{
    "Elasticsearch": {
        "Kibana": {
            "Case": [
                {
                    "id": "1234",
                    "title": "Suspicious login",
                    "description": "Multiple failed logins detected",
                    "owner": "securitySolution",
                    "severity": "medium",
                    "status": "open",
                    "created_at": "2024-01-15T10:00:00.000Z",
                    "connector": {"type": ".none"}
                }
            ]
        }
    }
}

Human Readable Output

Kibana Cases

Title ID Description Owner Severity Status Created at Connector type
Suspicious login 1234 Multiple failed logins detected securitySolution medium open 2024-01-15T10:00:00.000Z .none

es-kibana-case-alerts-list

Retrieves all alerts for a case.

Base Command

es-kibana-case-alerts-list

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
case_id The identifier for the case. To retrieve case IDs, use es-kibana-case-list. Required
limit Limit on the number of keys to return. Optional
offset Starting record index to begin retrieving records from. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.Case.Alert.id String The alert ID.
Elasticsearch.Kibana.Case.Alert.index String The alert index.
Elasticsearch.Kibana.Case.Alert.attached_at Date The date the alert was attached to the case.

Command Example

!es-kibana-case-alerts-list case_id="1234"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "Case": {
                "1234": {
                    "Alert": [
                        {
                            "id": "5678",
                            "index": ".alerts-security.alerts-default",
                            "attached_at": "2024-01-15T11:00:00.000Z"
                        }
                    ]
                }
            }
        }
    }
}

Human Readable Output

Alerts for Case 1234

Case id Attached at Alert id Index
1234 2024-01-15T11:00:00.000Z 5678 .alerts-security.alerts-default

es-kibana-case-comment-add

Add a case comment or alert.

Base Command

es-kibana-case-comment-add

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
case_id The identifier for the case. To retrieve case IDs, use es-kibana-case-list. Required
alert_id The alert identifiers. Required only when type is alert. Optional
index The alert indices. Required only when type is alert. Optional
owner The application that owns the cases: Stack Management, Observability, or Elastic Security. Required
rule_id The rule identifier. Optional
rule_name The rule name. Optional
type The type of comment. Required
comment The new comment. Required only when type is user. Maximum length is 30000. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.Case.id String The case ID.
Elasticsearch.Kibana.Case.comments.comment String The comment text.
Elasticsearch.Kibana.Case.comments.created_by.username String The user who created the comment.

Command Example

!es-kibana-case-comment-add case_id="1234" type="user" owner="securitySolution" comment="Investigated and confirmed malicious activity."

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "Case": {
                "id": "1234",
                "comments": [
                    {
                        "comment": "Investigated and confirmed malicious activity.",
                        "created_by": {"username": "analyst"}
                    }
                ]
            }
        }
    }
}

Human Readable Output

Kibana Case Comment Added

Case id Comment Created by
1234 Investigated and confirmed malicious activity. analyst

es-kibana-case-comment-update

Update a case comment or alert.

Base Command

es-kibana-case-comment-update

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
case_id The identifier for the case. To retrieve case IDs, use es-kibana-case-list. Required
comment_id The identifier for the comment. Optional
comment The new comment. Required only when type is user. Maximum length is 30000. Optional
alert_id The alert identifiers. Required only when type is alert. Optional
index The alert indices. Required only when type is alert. Optional
owner The application that owns the cases: Stack Management, Observability, or Elastic Security. Required
rule_id The rule identifier. Optional
rule_name The rule name. Optional
type The type of comment. Required
version The current comment version. To retrieve version values, use the get comments API. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.Case.id String The case ID.
Elasticsearch.Kibana.Case.comments.comment String The comment text.
Elasticsearch.Kibana.Case.comments.updated_by.username String The user who updated the comment.
Elasticsearch.Kibana.Case.comments.updated_at Date The date the comment was updated.

Command Example

!es-kibana-case-comment-update case_id="1234" comment_id="5678" type="user" owner="securitySolution" comment="Updated: confirmed false positive." version="WzEsMV0="

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "Case": {
                "id": "1234",
                "comments": [
                    {
                        "id": "5678",
                        "comment": "Updated: confirmed false positive.",
                        "updated_by": {"username": "analyst"},
                        "updated_at": "2024-01-15T12:00:00.000Z"
                    }
                ]
            }
        }
    }
}

Human Readable Output

Kibana Case Comment Updated

Case id Comment Updated by Updated at
1234 Updated: confirmed false positive. analyst 2024-01-15T12:00:00.000Z

es-kibana-case-comment-delete

Deletes all comments and alerts from a case.

Base Command

es-kibana-case-comment-delete

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
case_id The identifier for the case. To retrieve case IDs, use es-kibana-case-list. Required

Context Output

There is no context output for this command.

Command Example

!es-kibana-case-comment-delete case_id="1234"

Context Example

{}

Human Readable Output

The comments and alerts for the case 1234 have been successfully deleted.

es-kibana-case-file-attach

Attach a file to a case.

Base Command

es-kibana-case-file-attach

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
case_id The identifier for the case. To retrieve case IDs, use es-kibana-case-list. Required
entry_id Entry ID for the file that needs to be attached. Required
file_name The desired name of the file being attached to the case (without file extension). Optional

Context Output

Path Type Description
Elasticsearch.Kibana.Case.id String The case ID.
Elasticsearch.Kibana.Case.comments.updated_by.username String The user who attached the file.

Command Example

!es-kibana-case-file-attach case_id="1234" entry_id="1@1"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "Case": {
                "id": "1234",
                "comments": [
                    {
                        "updated_by": {"username": "analyst"}
                    }
                ]
            }
        }
    }
}

Human Readable Output

File successfully attached to case 1234.

es-kibana-endpoint-exception-list-item-create

Create an Elastic Endpoint exception list item, and associate it with the Elastic Endpoint exception list.

Base Command

es-kibana-endpoint-exception-list-item-create

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
description Describes the exception list. Optional
entries_field A string that does not contain only whitespace characters. Optional
entries_list_id Value list’s identifier. Optional
entries_list_type Specifies the Elasticsearch data type of excludes the list container holds. Optional
entries_operator The exception item entry operator. Optional
entries_type The exception item entry type. Optional
item_id Human readable string identifier, e.g. trusted-linux-processes. Optional
meta Additional properties are allowed (JSON object). Optional
name Exception list name. Minimum length is 1. Optional
os_types Use this field to specify the operating system. Optional
tags String array containing words and phrases to help categorize exception items. Optional
entry_id Entry ID for the file containing the full request JSON. If provided, other parameters won’t be considered. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.EndpointExceptionListItem.id String The exception list item ID.
Elasticsearch.Kibana.EndpointExceptionListItem.item_id String The exception list item human-readable ID.
Elasticsearch.Kibana.EndpointExceptionListItem.list_id String The exception list ID.
Elasticsearch.Kibana.EndpointExceptionListItem.name String The exception list item name.
Elasticsearch.Kibana.EndpointExceptionListItem.description String The exception list item description.
Elasticsearch.Kibana.EndpointExceptionListItem.created_at Date The creation date of the exception list item.

Command Example

!es-kibana-endpoint-exception-list-item-create name="Trusted process" description="Allow known safe process" entries_field="process.name" entries_type="match" entries_operator="included" entries_value="safe_process.exe" os_types="windows"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "EndpointExceptionListItem": {
                "id": "1234",
                "item_id": "trusted-process-1",
                "list_id": "endpoint_list",
                "name": "Trusted process",
                "description": "Allow known safe process",
                "created_at": "2024-01-15T10:00:00.000Z"
            }
        }
    }
}

Human Readable Output

Kibana Endpoint Exception List Item

ID Item ID List ID Name Description Creation date
1234 trusted-process-1 endpoint_list Trusted process Allow known safe process 2024-01-15T10:00:00.000Z

es-kibana-endpoint-exception-list-item-update

Update an Elastic Endpoint exception list item.

Base Command

es-kibana-endpoint-exception-list-item-update

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
exception_list_item_id Exception’s identifier. Optional
item_id Human readable string identifier, e.g. trusted-linux-processes. Optional
_version The version id, normally returned by the API when the item is retrieved. Optional
description Describes the exception list. Optional
entries_field A string that does not contain only whitespace characters. Optional
entries_list_id Value list’s identifier. Optional
entries_list_type Specifies the Elasticsearch data type of excludes the list container holds. Optional
entries_operator The exception item entry operator. Optional
entries_type The exception item entry type. Optional
entries_value A string that does not contain only whitespace characters. Optional
meta Additional properties are allowed (JSON object). Optional
name Exception list name. Minimum length is 1. Optional
os_types Use this field to specify the operating system. Optional
tags String array containing words and phrases to help categorize exception items. Optional
entry_id Entry ID for the file containing the full request JSON. If provided, other parameters won’t be considered. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.EndpointExceptionListItem.id String The exception list item ID.
Elasticsearch.Kibana.EndpointExceptionListItem.item_id String The exception list item human-readable ID.
Elasticsearch.Kibana.EndpointExceptionListItem.list_id String The exception list ID.
Elasticsearch.Kibana.EndpointExceptionListItem.name String The exception list item name.
Elasticsearch.Kibana.EndpointExceptionListItem.description String The exception list item description.
Elasticsearch.Kibana.EndpointExceptionListItem.created_at Date The creation date of the exception list item.

Command Example

!es-kibana-endpoint-exception-list-item-update item_id="trusted-process-1" name="Trusted process updated" description="Updated description" entries_field="process.name" entries_type="match" entries_operator="included" entries_value="safe_process.exe"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "EndpointExceptionListItem": {
                "id": "1234",
                "item_id": "trusted-process-1",
                "list_id": "endpoint_list",
                "name": "Trusted process updated",
                "description": "Updated description",
                "created_at": "2024-01-15T10:00:00.000Z"
            }
        }
    }
}

Human Readable Output

Kibana Endpoint Exception List Item

ID Item ID List ID Name Description Creation date
1234 trusted-process-1 endpoint_list Trusted process updated Updated description 2024-01-15T10:00:00.000Z

es-kibana-endpoint-exception-list-item-delete

Delete an Elastic Endpoint exception list item.

Base Command

es-kibana-endpoint-exception-list-item-delete

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
item_id Either id or item_id must be specified. Required

Context Output

There is no context output for this command.

Command Example

!es-kibana-endpoint-exception-list-item-delete item_id="trusted-process-1"

Context Example

{}

Human Readable Output

The item trusted-process-1 has been successfully deleted.

es-kibana-endpoint-exception-list-item-list

Retrieves Elastic Endpoint exception list items.

Base Command

es-kibana-endpoint-exception-list-item-list

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
filter Filters the returned results according to the value of the specified field, using the <field name>:<field value> syntax. Optional
item_id Either id or item_id must be specified. Optional
sort_field Determines which field is used to sort the results. Optional
sort_order Determines the sort order. Optional
page The page number to return. Minimum value is 0. Optional
size The number of exception list items to return per page. Minimum value is 0. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.EndpointExceptionListItem.id String The exception list item ID.
Elasticsearch.Kibana.EndpointExceptionListItem.item_id String The exception list item human-readable ID.
Elasticsearch.Kibana.EndpointExceptionListItem.list_id String The exception list ID.
Elasticsearch.Kibana.EndpointExceptionListItem.name String The exception list item name.
Elasticsearch.Kibana.EndpointExceptionListItem.description String The exception list item description.
Elasticsearch.Kibana.EndpointExceptionListItem.created_at Date The creation date of the exception list item.

Command Example


#### Context Example

```json
{
    "Elasticsearch": {
        "Kibana": {
            "EndpointExceptionListItem": [
                {
                    "id": "1234",
                    "item_id": "trusted-process-1",
                    "list_id": "endpoint_list",
                    "name": "Trusted process",
                    "description": "Allow known safe process",
                    "created_at": "2024-01-15T10:00:00.000Z"
                }
            ]
        }
    }
}

Human Readable Output

Kibana Endpoint Exception List Items

ID Item ID List ID Name Description Creation date
1234 trusted-process-1 endpoint_list Trusted process Allow known safe process 2024-01-15T10:00:00.000Z

es-kibana-exception-list-list

Get a list of all exception list containers.

Base Command

es-kibana-exception-list-list

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
exception_list_id Exception list’s identifier. Either id or list_id must be specified. Optional
list_id Human readable exception list string identifier. Either id or list_id must be specified. Optional
filter Filters the returned results according to the value of the specified field. Optional
namespace_type Determines whether the returned containers are Kibana associated with a Kibana space or available in all spaces. Optional
sort_field Determines which field is used to sort the results. Optional
sort_order Determines the sort order. Optional
page The page number to return. Minimum value is 1. Optional
size The number of exception lists to return per page. Minimum value is 1. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.ExceptionList.id String The exception list ID.
Elasticsearch.Kibana.ExceptionList.list_id String The human-readable exception list ID.
Elasticsearch.Kibana.ExceptionList.name String The exception list name.
Elasticsearch.Kibana.ExceptionList.description String The exception list description.
Elasticsearch.Kibana.ExceptionList.created_at Date The creation date of the exception list.

Command Example


#### Context Example

```json
{
    "Elasticsearch": {
        "Kibana": {
            "ExceptionList": [
                {
                    "id": "1234",
                    "list_id": "my-exception-list",
                    "name": "My Exception List",
                    "description": "Exceptions for trusted processes",
                    "created_at": "2024-01-15T10:00:00.000Z"
                }
            ]
        }
    }
}

Human Readable Output

Kibana Exception Lists

Exception list ID List ID Name Description Creation date
1234 my-exception-list My Exception List Exceptions for trusted processes 2024-01-15T10:00:00.000Z

es-kibana-exception-list-create

Create an exception list.

Base Command

es-kibana-exception-list-create

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
description Describes the exception list. Optional
list_id The exception list’s human-readable string identifier. Optional
meta Placeholder for metadata about the list container (JSON object). Optional
name The name of the exception list. Optional
namespace_type Determines whether the returned containers are Kibana associated with a Kibana space or available in all spaces. Optional
os_types Use this field to specify the operating system. Only enter one value. Optional
tags String array containing words and phrases to help categorize exception containers. Optional
type The type of exception list to be created. Required
version The document version automatically increased on updates. Optional
entry_id Entry ID for the file containing the request JSON. If provided, other parameters won’t be considered. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.ExceptionList.id String The exception list ID.
Elasticsearch.Kibana.ExceptionList.list_id String The human-readable exception list ID.
Elasticsearch.Kibana.ExceptionList.name String The exception list name.
Elasticsearch.Kibana.ExceptionList.description String The exception list description.
Elasticsearch.Kibana.ExceptionList.created_at Date The creation date of the exception list.

Command Example

!es-kibana-exception-list-create type="detection" name="My Exception List" description="Exceptions for trusted processes" list_id="my-exception-list"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "ExceptionList": {
                "id": "1234",
                "list_id": "my-exception-list",
                "name": "My Exception List",
                "description": "Exceptions for trusted processes",
                "created_at": "2024-01-15T10:00:00.000Z"
            }
        }
    }
}

Human Readable Output

Kibana Exception List

Exception list ID List ID Name Description Creation date
1234 my-exception-list My Exception List Exceptions for trusted processes 2024-01-15T10:00:00.000Z

es-kibana-exception-list-update

Updates an existing exception list.

Base Command

es-kibana-exception-list-update

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
_version The version id, normally returned by the API when the item was retrieved. Optional
description Describes the exception list. Required
exception_list_id Exception list’s identifier. Optional
list_id The exception list’s human-readable string identifier. Optional
meta Placeholder for metadata about the list container (JSON object). Optional
name The name of the exception list. Required
namespace_type Determines whether the returned containers are Kibana associated with a Kibana space or available in all spaces. Optional
os_types Use this field to specify the operating system. Only enter one value. Optional
tags String array containing words and phrases to help categorize exception containers. Optional
type The type of exception list to be created. Required
version The document version automatically increased on updates. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.ExceptionList.id String The exception list ID.
Elasticsearch.Kibana.ExceptionList.list_id String The human-readable exception list ID.
Elasticsearch.Kibana.ExceptionList.name String The exception list name.
Elasticsearch.Kibana.ExceptionList.description String The exception list description.
Elasticsearch.Kibana.ExceptionList.created_at Date The creation date of the exception list.

Command Example

!es-kibana-exception-list-update exception_list_id="1234" name="My Exception List Updated" description="Updated description" type="detection"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "ExceptionList": {
                "id": "1234",
                "list_id": "my-exception-list",
                "name": "My Exception List Updated",
                "description": "Updated description",
                "created_at": "2024-01-15T10:00:00.000Z"
            }
        }
    }
}

Human Readable Output

Kibana Exception List

Exception list ID List ID Name Description Creation date
1234 my-exception-list My Exception List Updated Updated description 2024-01-15T10:00:00.000Z

es-kibana-exception-list-delete

Delete an exception list using the id or list_id field.

Base Command

es-kibana-exception-list-delete

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
exception_list_id Exception list’s identifier. Either id or list_id must be specified. Optional
list_id Human readable exception list string identifier. Either id or list_id must be specified. Optional
namespace_type Determines whether the returned containers are Kibana associated with a Kibana space or available in all spaces. Optional

Context Output

There is no context output for this command.

Command Example

!es-kibana-exception-list-delete list_id="my-exception-list"

Context Example

{}

Human Readable Output

The exception list my-exception-list has been successfully deleted.

es-kibana-exception-list-item-list

Get a list of all exception list items in the specified list.

Base Command

es-kibana-exception-list-item-list

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
exception_list_item_id Exception list item’s identifier. Either id or item_id must be specified. Optional
item_id Human readable exception item string identifier. Either id or item_id must be specified. Optional
exception_list_id The list_ids of the items to fetch. Optional
filter Filters the returned results according to the value of the specified field, using the <field name>:<field value> syntax. Optional
namespace_type Determines whether the returned containers are Kibana associated with a Kibana space or available in all spaces. Optional
search Free-text search term applied to exception list item fields. Optional
sort_field Determines which field is used to sort the results. Optional
sort_order Determines the sort order. Optional
page The page number to return. Minimum value is 0. Optional
size The number of exception list items to return per page. Minimum value is 0. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.ExceptionListItem.id String The exception list item ID.
Elasticsearch.Kibana.ExceptionListItem.item_id String The exception list item human-readable ID.
Elasticsearch.Kibana.ExceptionListItem.list_id String The exception list ID.
Elasticsearch.Kibana.ExceptionListItem.name String The exception list item name.
Elasticsearch.Kibana.ExceptionListItem.description String The exception list item description.
Elasticsearch.Kibana.ExceptionListItem.created_at Date The creation date of the exception list item.

Command Example

!es-kibana-exception-list-item-list exception_list_id="my-exception-list"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "ExceptionListItem": [
                {
                    "id": "1234",
                    "item_id": "my-item-1",
                    "list_id": "my-exception-list",
                    "name": "Trusted IP",
                    "description": "Known safe IP address",
                    "created_at": "2024-01-15T10:00:00.000Z"
                }
            ]
        }
    }
}

Human Readable Output

Kibana Exception List Items

ID Item ID List ID Name Description Creation date
1234 my-item-1 my-exception-list Trusted IP Known safe IP address 2024-01-15T10:00:00.000Z

es-kibana-exception-list-item-create

Create an exception item and associate it with the specified exception list.

Base Command

es-kibana-exception-list-item-create

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
comment A string that does not contain only whitespace characters. Optional
description Describes the exception list. Optional
expire_time The exception item’s expiration date, in ISO format. Optional
item_id Human readable string identifier, e.g. trusted-linux-processes. Optional
meta Additional properties are allowed (JSON object). Optional
name Exception list name. Optional
namespace_type Determines whether the returned containers are Kibana associated with a Kibana space or available in all spaces. Optional
type The type of exception list to be created. Optional
entries_field A string that does not contain only whitespace characters. Optional
entries_operator The exception item entry operator. Optional
entries_type The exception item entry type. Optional
entries_value The exception item entry value. Optional
entries_list_id Value list’s identifier. Optional
entries_list_type Specifies the Elasticsearch data type of excludes the list container holds. Optional
list_id The exception list’s human-readable string identifier. Optional
os_types Use this field to specify the operating system. Only enter one value. Optional
tags String array containing words and phrases to help categorize exception containers. Optional
entry_id Entry ID for the file containing the request JSON. If provided, other parameters won’t be considered. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.ExceptionListItem.id String The exception list item ID.
Elasticsearch.Kibana.ExceptionListItem.item_id String The exception list item human-readable ID.
Elasticsearch.Kibana.ExceptionListItem.list_id String The exception list ID.
Elasticsearch.Kibana.ExceptionListItem.name String The exception list item name.
Elasticsearch.Kibana.ExceptionListItem.description String The exception list item description.
Elasticsearch.Kibana.ExceptionListItem.created_at Date The creation date of the exception list item.

Command Example

!es-kibana-exception-list-item-create list_id="my-exception-list" name="Trusted IP" description="Known safe IP address" entries_field="source.ip" entries_type="match" entries_operator="included" entries_value="192.168.1.1"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "ExceptionListItem": {
                "id": "1234",
                "item_id": "my-item-1",
                "list_id": "my-exception-list",
                "name": "Trusted IP",
                "description": "Known safe IP address",
                "created_at": "2024-01-15T10:00:00.000Z"
            }
        }
    }
}

Human Readable Output

Kibana Exception List Item

ID Item ID List ID Name Description Creation date
1234 my-item-1 my-exception-list Trusted IP Known safe IP address 2024-01-15T10:00:00.000Z

es-kibana-exception-item-list-update

Updates an existing exception list item.

Base Command

es-kibana-exception-item-list-update

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
_version The version ID, normally returned by the API when the item is retrieved. Optional
comment A string that does not contain only whitespace characters. Optional
comment_id A string that does not contain only whitespace characters. Optional
description Describes the exception list. Optional
expire_time The exception item’s expiration date, in ISO format. Optional
exception_list_item_id Exception’s identifier. Optional
item_id Human readable string identifier, e.g. trusted-linux-processes. Optional
meta Additional properties are allowed (JSON object). Optional
name Exception list name. Optional
namespace_type Determines whether the returned containers are Kibana associated with a Kibana space or available in all spaces. Optional
type The type of exception list to be created. Optional
entries_field A string that does not contain only whitespace characters. Optional
entries_operator The exception item entry operator. Optional
entries_type The exception item entry type. Optional
entries_value The exception item entry value. Optional
entries_list_id Value list’s identifier. Optional
entries_list_type Specifies the Elasticsearch data type of excludes the list container holds. Optional
list_id The exception list’s human-readable string identifier. Optional
os_types Use this field to specify the operating system. Only enter one value. Optional
tags String array containing words and phrases to help categorize exception containers. Optional
entry_id Entry ID for the file containing the request JSON. If provided, other parameters won’t be considered. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.ExceptionListItem.id String The exception list item ID.
Elasticsearch.Kibana.ExceptionListItem.item_id String The exception list item human-readable ID.
Elasticsearch.Kibana.ExceptionListItem.list_id String The exception list ID.
Elasticsearch.Kibana.ExceptionListItem.name String The exception list item name.
Elasticsearch.Kibana.ExceptionListItem.description String The exception list item description.
Elasticsearch.Kibana.ExceptionListItem.updated_at Date The update date of the exception list item.

Command Example

!es-kibana-exception-item-list-update item_id="my-item-1" name="Trusted IP updated" description="Updated safe IP" entries_field="source.ip" entries_type="match" entries_operator="included" entries_value="192.168.1.1"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "ExceptionListItem": {
                "id": "1234",
                "item_id": "my-item-1",
                "list_id": "my-exception-list",
                "name": "Trusted IP updated",
                "description": "Updated safe IP",
                "updated_at": "2024-01-15T12:00:00.000Z"
            }
        }
    }
}

Human Readable Output

Kibana Exception List Item Updated

Exception list item ID Item Id List ID Name Description Update date
1234 my-item-1 my-exception-list Trusted IP updated Updated safe IP 2024-01-15T12:00:00.000Z

es-kibana-exception-list-item-delete

Deletes an exception list item.

Base Command

es-kibana-exception-list-item-delete

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
exception_list_item_id Exception list item’s identifier. Either id or item_id must be specified. Optional
item_id Human readable exception item string identifier. Either id or item_id must be specified. Optional
namespace_type Determines whether the returned containers are Kibana associated with a Kibana space or available in all spaces. Optional

Context Output

There is no context output for this command.

Command Example

!es-kibana-exception-list-item-delete item_id="my-item-1"

Context Example

{}

Human Readable Output

The exception list item my-item-1 has been successfully deleted.

es-kibana-value-lists-list

Retrieves details of a value list (the list container).

Base Command

es-kibana-value-lists-list

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
value_list_id Value list identifier (id) returned when the list was created. Optional
filter Filters the returned results according to the value of the specified field, using the <field name>:<field value> syntax. Optional
cursor Returns the lists that come after the last lists returned in the previous call. Optional
sort_field Determines which field is used to sort the results. Optional
sort_order Determines the sort order. Optional
page The page number to return. Optional
size The number of value lists to return per page. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.ValueList.id String The value list ID.
Elasticsearch.Kibana.ValueList.name String The value list name.
Elasticsearch.Kibana.ValueList.description String The value list description.
Elasticsearch.Kibana.ValueList.created_at Date The creation date of the value list.

Command Example


#### Context Example

```json
{
    "Elasticsearch": {
        "Kibana": {
            "ValueList": [
                {
                    "id": "1234",
                    "name": "trusted-ips",
                    "description": "List of trusted IP addresses",
                    "created_at": "2024-01-15T10:00:00.000Z"
                }
            ]
        }
    }
}

Human Readable Output

Kibana Value Lists

Value list ID Name Description Creation date
1234 trusted-ips List of trusted IP addresses 2024-01-15T10:00:00.000Z

es-kibana-value-list-item-get

Retrieves value list items.

Base Command

es-kibana-value-list-item-get

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
value_list_item_id Value list item identifier. Required if list_id and value are not specified. Optional
value_list_id Parent value list’s id to page through items for. Optional
value The value used to evaluate exceptions. Required if id is not specified. Optional
filter Filters the returned results according to the value of the specified field, using the <field name>:<field value> syntax. Optional
cursor Opaque cursor returned in a previous response; pass it to continue listing from the next page. Optional
sort_field Determines which field is used to sort the results. Optional
sort_order Determines the sort order. Optional
page The page number to return. Optional
size The number of list items to return per page. Optional

Context Output

Path Type Description
Elasticsearch.Kibana.ValueListItem.id String The value list item ID.
Elasticsearch.Kibana.ValueListItem.list_id String The value list ID.
Elasticsearch.Kibana.ValueListItem.name String The value list item name.
Elasticsearch.Kibana.ValueListItem.description String The value list item description.
Elasticsearch.Kibana.ValueListItem.created_at Date The creation date of the value list item.

Command Example

!es-kibana-value-list-item-get value_list_id="1234"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "ValueListItem": [
                {
                    "id": "5678",
                    "list_id": "1234",
                    "value": "192.168.1.1",
                    "created_at": "2024-01-15T10:00:00.000Z"
                }
            ]
        }
    }
}

Human Readable Output

Kibana Value List Items

Value list item ID List ID Value Creation date
5678 1234 192.168.1.1 2024-01-15T10:00:00.000Z

es-kibana-value-list-item-create

Adds a new item to a value list.

Base Command

es-kibana-value-list-item-create

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
value_list_id Value list’s identifier. Required
meta Placeholder for metadata about the value list item (JSON object). Example {“source”:”threatfeed”,”priority”:3,”active”:true}. Optional
refresh Determines when changes made by the request are made visible to search. Optional
value The value used to evaluate exceptions. Required

Context Output

Path Type Description
Elasticsearch.Kibana.ValueListItem.id String The value list item ID.
Elasticsearch.Kibana.ValueListItem.list_id String The value list ID.
Elasticsearch.Kibana.ValueListItem.name String The value list item name.
Elasticsearch.Kibana.ValueListItem.description String The value list item description.
Elasticsearch.Kibana.ValueListItem.created_at Date The creation date of the value list item.

Command Example

!es-kibana-value-list-item-create value_list_id="1234" value="10.0.0.1"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "ValueListItem": {
                "id": "5678",
                "list_id": "1234",
                "value": "10.0.0.1",
                "created_at": "2024-01-15T10:00:00.000Z"
            }
        }
    }
}

Human Readable Output

Kibana Value List Item

Value list item ID List ID Value Creation date
5678 1234 10.0.0.1 2024-01-15T10:00:00.000Z

es-kibana-value-list-item-update

Updates an existing value list item.

Base Command

es-kibana-value-list-item-update

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
_version The version id, normally returned by the API when the document is retrieved. Optional
value_list_item_id Value list item identifier. Required
meta Placeholder for metadata about the value list item (JSON object). Example {“source”:”threatfeed”,”priority”:3,”active”:true}. Optional
value The value used to evaluate exceptions. Required

Context Output

Path Type Description
Elasticsearch.Kibana.ValueListItem.id String The value list item ID.
Elasticsearch.Kibana.ValueListItem.list_id String The value list ID.
Elasticsearch.Kibana.ValueListItem.name String The value list item name.
Elasticsearch.Kibana.ValueListItem.description String The value list item description.
Elasticsearch.Kibana.ValueListItem.created_at Date The creation date of the value list item.

Command Example

!es-kibana-value-list-item-update value_list_item_id="5678" value="10.0.0.2"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "ValueListItem": {
                "id": "5678",
                "list_id": "1234",
                "value": "10.0.0.2",
                "created_at": "2024-01-15T10:00:00.000Z"
            }
        }
    }
}

Human Readable Output

Kibana Value List Item

Value list item ID List ID Value Creation date
5678 1234 10.0.0.2 2024-01-15T10:00:00.000Z

es-kibana-value-list-item-delete

Deletes a value list item.

Base Command

es-kibana-value-list-item-delete

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
value_list_item_id Value list item’s identifier. Required if list_id and value are not specified. Optional
value_list_id Value list’s identifier. Optional
value The value used to evaluate exceptions. Required if id is not specified. Optional
refresh Determines when changes made by the request are made visible to search. Optional

Context Output

There is no context output for this command.

Command Example

!es-kibana-value-list-item-delete value_list_item_id="5678"

Context Example

{}

Human Readable Output

The value list item 5678 has been successfully deleted.

es-kibana-value-list-item-export

Exports all items of a value list as a file (returned to the War Room).

Base Command

es-kibana-value-list-item-export

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
value_list_id Value list’s ID to export. Optional

Context Output

There is no context output for this command.

Command Example

!es-kibana-value-list-item-export value_list_id="1234"

Context Example

{}

Human Readable Output

The value list 1234 has been exported successfully.

es-kibana-value-list-item-import

Import value list items from a TXT or CSV file.

Base Command

es-kibana-value-list-item-import

Input

Argument Name Description Required
space_id Refer to https://www.elastic.co/docs/deploy-manage/manage-spaces for more information. Optional
value_list_id List’s ID to import. Optional
type Type of the importing list. Optional
refresh Determines when changes made by the request are made visible to search. Optional
entry_id Entry ID for the file containing the items to import. Required

Context Output

Path Type Description
Elasticsearch.Kibana.ValueListItem.id String The value list item ID.
Elasticsearch.Kibana.ValueListItem.list_id String The value list ID.
Elasticsearch.Kibana.ValueListItem.name String The value list item name.
Elasticsearch.Kibana.ValueListItem.description String The value list item description.
Elasticsearch.Kibana.ValueListItem.created_at Date The creation date of the value list item.

Command Example

!es-kibana-value-list-item-import value_list_id="1234" entry_id="1@1"

Context Example

{
    "Elasticsearch": {
        "Kibana": {
            "ValueListItem": {
                "id": "5678",
                "list_id": "1234",
                "value": "10.0.0.1",
                "created_at": "2024-01-15T10:00:00.000Z"
            }
        }
    }
}

Human Readable Output

Kibana Value List Item

Value list item ID List ID Value Creation date
5678 1234 10.0.0.1 2024-01-15T10:00:00.000Z

<~PLATFORM>

License Requirements

The following configuration parameters require one of these licenses: Cortex XSIAM or Agentix:

  • Fetch incidents

</~PLATFORM>

Configuration parameters

  • url — Server URL
  • kibana_url — Kibana Server URL
  • auth_type — Authorization type (required)
  • api_key_auth_credentials — API key ID
  • credentials — Username
  • insecure — Trust any certificate (not secure)
  • proxy — Use system proxy settings
  • client_type — Client type
  • fetch_index — Index to fetch incidents from
  • fetch_query — Query String
  • fetch_fields — Fields to Fetch
  • fetch_time_field — Index time field
  • raw_query — Raw Query
  • time_method — Time field type
  • map_labels — Map JSON fields into labels
  • fetch_time — First fetch timestamp
  • fetch_size — The maximum number of results per fetch
  • timeout — Request timeout (in seconds).
  • incidentType — Incident type
  • isFetch — Fetch incidents
  • incidentFetchInterval — Incidents Fetch Interval
  • space_id — Space ID
  • fetch_incident_type — Fetch incident types
  • mirror_direction — Incident Mirroring Direction
  • fetch_severity — Fetch cases by Severity
  • fetch_status — Fetch cases by Status
  • fetch_alerts_for_case — Fetch alerts for case
  • close_incident — Close Mirrored XSOAR Incident
  • close_elastic_incident — Close Mirrored Elasticsearch Incident

Commands (48)

  • es-eql-search

    Search using EQL query.

  • es-esql-search

    Search using ES|QL query. (Elastic >=8.11).

  • es-get-indices-statistics

    Returns Elasticsearch indices statistics and information. This command is not supported for client type OpenSearch.

  • es-index

    Indexes a document into an Elasticsearch index.

  • es-integration-health-check

    Returns the health status of the integration. This commmand should be used for debugging purposes.

  • es-kibana-alerting-health-get

    Retrieves the health of the Kibana alerting framework.

  • es-kibana-case-alerts-list

    Retrieves all alerts for a case.

  • es-kibana-case-comment-add

    Add a case comment or alert.

  • es-kibana-case-comment-delete

    Deletes all comments and alerts from a case.

  • es-kibana-case-comment-update

    Update a case comment or alert.

  • es-kibana-case-create

    Creates a new case in Kibana.

  • es-kibana-case-delete

    Deletes one or more cases by ID.

  • es-kibana-case-file-attach

    Attach a file to a case.

  • es-kibana-case-list

    Retrieves the details of Kibana cases.

  • es-kibana-case-update

    Update cases in Kibana.

  • es-kibana-detection-alert-status-set

    Set the status of one or more detection alerts.

  • es-kibana-endpoint-exception-list-item-create

    Create an Elastic Endpoint exception list item, and associate it with the Elastic Endpoint exception list.

  • es-kibana-endpoint-exception-list-item-delete

    Delete an Elastic Endpoint exception list item.

  • es-kibana-endpoint-exception-list-item-list

    Retrieves Elastic Endpoint exception list items.

  • es-kibana-endpoint-exception-list-item-update

    Update an Elastic Endpoint exception list item.

  • es-kibana-exception-item-list-update

    Updates an existing exception list item.

  • es-kibana-exception-list-create

    Create an exception list.

  • es-kibana-exception-list-delete

    Delete an exception list using the id or list_id field.

  • es-kibana-exception-list-item-create

    Create an exception item and associate it with the specified exception list.

  • es-kibana-exception-list-item-delete

    Deletes an exception list item.

  • es-kibana-exception-list-item-list

    Get a list of all exception list items in the specified list.

  • es-kibana-exception-list-list

    Get a list of all exception list containers.

  • es-kibana-exception-list-update

    Updates an existing exception list.

  • es-kibana-rule-alert-mute

    Mute an alert, or mute all alerts for a rule.

  • es-kibana-rule-alert-unmute

    Unmute an alert, or unmute all alerts for a rule.

  • es-kibana-rule-disable

    Disable a rule.

  • es-kibana-rule-enable

    Enable a rule.

  • es-kibana-rule-list

    Retrieves information about rules.

  • es-kibana-rule-types-list

    Retrieves all rule types available in Kibana.

  • es-kibana-rule-update

    Update a rule.

  • es-kibana-value-list-item-create

    Adds a new item to a value list.

  • es-kibana-value-list-item-delete

    Deletes a value list item.

  • es-kibana-value-list-item-export

    Exports all items of a value list as a file (returned to the War Room).

  • es-kibana-value-list-item-get

    Retrieves value list items.

  • es-kibana-value-list-item-import

    Import value list items from a TXT or CSV file.

  • es-kibana-value-list-item-update

    Updates an existing value list item.

  • es-kibana-value-lists-list

    Retrieves details of a value list (the list container).

  • es-search

    Queries an index.

  • get-mapping-fields

    Returns the schema of the index to fetch from. This commmand should be used for debugging purposes.

  • get-modified-remote-data

    Gets the list of incidents that were modified since the last update time. This command is used for mirroring in and should be used for debugging purposes.

  • get-remote-data

    Gets remote data of a specific incident. This command is used for mirroring in and should be used for debugging purposes.

  • search

    Searches an index.

  • update-remote-system

    Updates the remote system with local incident changes. This command is used for mirroring out and should be used for debugging purposes.

Source

Elasticsearch_v2.py Elasticsearch_v2_test.py Elasticsearch_v2.yml Elasticsearch_v2_description.md
Download History Raw 
import importlib
import unittest
from datetime import datetime, timedelta, UTC
from unittest import mock
from unittest.mock import patch, MagicMock

import dateparser
import demistomock as demisto
import Elasticsearch_v2
import json
import pytest
import requests

"""MOCKED RESPONSES"""

ES_V6_RESPONSE = {
    "took": 1,
    "timed_out": False,
    "_shards": {"total": 5, "successful": 5, "skipped": 0, "failed": 0},
    "hits": {
        "total": 17,
        "max_score": 1.3862944,
        "hits": [
            {
                "_index": "users",
                "_type": "_doc",
                "_id": "123",
                "_score": 1.3862944,
                "_source": {"Date": "2019-08-29T14:45:00.123Z"},
            },
            {
                "_index": "users",
                "_type": "_doc",
                "_id": "456",
                "_score": 0.9517491,
                "_source": {"Date": "2019-08-29T14:46:00.123456Z"},
            },
        ],
    },
}

ES_V7_RESPONSE = {
    "took": 1,
    "timed_out": False,
    "_shards": {"total": 1, "successful": 1, "skipped": 0, "failed": 0},
    "hits": {
        "total": {"value": 9, "relation": "eq"},
        "max_score": 0.6814878,
        "hits": [
            {
                "_index": "customer",
                "_type": "doc",
                "_id": "123",
                "_score": 0.6814878,
                "_source": {"Date": "2019-08-27T18:00:00Z"},
            },
            {
                "_index": "customer",
                "_type": "doc",
                "_id": "456",
                "_score": 0.6814878,
                "_source": {"Date": "2019-08-27T18:01:25.343212Z"},
            },
        ],
    },
}

ES_V8_RESPONSE = {
    "took": 8,
    "timed_out": False,
    "_shards": {"total": 1, "successful": 1, "skipped": 0, "failed": 0},
    "hits": {
        "total": {"value": 9, "relation": "eq"},
        "max_score": 0.8,
        "hits": [
            {
                "_index": "customer",
                "_type": "doc",
                "_id": "888",
                "_score": 0.8,
                "_source": {"Date": "2024-08-27T18:00:00Z"},
            },
            {
                "_index": "customer",
                "_type": "doc",
                "_id": "999",
                "_score": 0.79,
                "_source": {"Date": "2024-08-27T18:01:25.343212Z"},
            },
        ],
    },
}


MOCK_ES7_SEARCH_CONTEXT = str(
    {
        "Server": "",
        "Index": "customer",
        "Query": "check",
        "Page": 0,
        "Size": 2,
        "total": {"value": 9, "relation": "eq"},
        "max_score": 0.6814878,
        "took": 1,
        "timed_out": False,
        "Results": [
            {
                "_index": "customer",
                "_type": "doc",
                "_id": "123",
                "_score": 0.6814878,
                "_source": {"Date": "2019-08-27T18:00:00Z"},
            },
            {
                "_index": "customer",
                "_type": "doc",
                "_id": "456",
                "_score": 0.6814878,
                "_source": {"Date": "2019-08-27T18:01:25.343212Z"},
            },
        ],
    }
)

MOCK_ES7_HIT_CONTEXT = str(
    [
        {"_index": "customer", "_id": "123", "_type": "doc", "_score": 0.6814878, "Date": "2019-08-27T18:00:00Z"},
        {"_index": "customer", "_id": "456", "_type": "doc", "_score": 0.6814878, "Date": "2019-08-27T18:01:25.343212Z"},
    ]
)

MOCK_ES6_SEARCH_CONTEXT = str(
    {
        "Server": "",
        "Index": "users",
        "Query": "incident",
        "Page": 0,
        "Size": 2,
        "total": {"value": 17},
        "max_score": 1.3862944,
        "took": 1,
        "timed_out": False,
        "Results": [
            {
                "_index": "users",
                "_type": "_doc",
                "_id": "123",
                "_score": 1.3862944,
                "_source": {"Date": "2019-08-29T14:45:00.123Z"},
            },
            {
                "_index": "users",
                "_type": "_doc",
                "_id": "456",
                "_score": 0.9517491,
                "_source": {"Date": "2019-08-29T14:46:00.123456Z"},
            },
        ],
    }
)

MOCK_ES6_HIT_CONTEXT = str(
    [
        {"_index": "users", "_id": "123", "_type": "_doc", "_score": 1.3862944, "Date": "2019-08-29T14:45:00.123Z"},
        {"_index": "users", "_id": "456", "_type": "_doc", "_score": 0.9517491, "Date": "2019-08-29T14:46:00.123456Z"},
    ]
)

MOCK_ES7_INCIDENTS = str(
    [
        {
            "name": "Elasticsearch: Index: customer, ID: 123",
            "rawJSON": "{"
            '"_index": "customer", '
            '"_type": "doc", '
            '"_id": "123", '
            '"_score": 0.6814878, '
            '"_source": {"Date": "2019-08-27T18:00:00Z"}'
            "}",
            "occurred": "2019-08-27T18:00:00Z",
            "dbotMirrorId": "123",
            "labels": [{"type": "Date", "value": "2019-08-27T18:00:00Z"}],
        },
        {
            "name": "Elasticsearch: Index: customer, ID: 456",
            "rawJSON": "{"
            '"_index": "customer", '
            '"_type": "doc", '
            '"_id": "456", '
            '"_score": 0.6814878, '
            '"_source": {"Date": "2019-08-27T18:01:25.343212Z"}'
            "}",
            "occurred": "2019-08-27T18:01:25Z",
            "dbotMirrorId": "456",
            "labels": [{"type": "Date", "value": "2019-08-27T18:01:25.343212Z"}],
        },
    ]
)

MOCK_ES7_INCIDENTS_WITHOUT_LABELS = str(
    [
        {
            "name": "Elasticsearch: Index: customer, ID: 123",
            "rawJSON": "{"
            '"_index": "customer", '
            '"_type": "doc", '
            '"_id": "123", '
            '"_score": 0.6814878, '
            '"_source": {"Date": "2019-08-27T18:00:00Z"}'
            "}",
            "occurred": "2019-08-27T18:00:00Z",
            "dbotMirrorId": "123",
        },
        {
            "name": "Elasticsearch: Index: customer, ID: 456",
            "rawJSON": "{"
            '"_index": "customer", '
            '"_type": "doc", '
            '"_id": "456", '
            '"_score": 0.6814878, '
            '"_source": {"Date": "2019-08-27T18:01:25.343212Z"}'
            "}",
            "occurred": "2019-08-27T18:01:25Z",
            "dbotMirrorId": "456",
        },
    ]
)

MOCK_ES6_INCIDETNS = str(
    [
        {
            "name": "Elasticsearch: Index: users, ID: 123",
            "rawJSON": "{"
            '"_index": "users", '
            '"_type": "_doc", '
            '"_id": "123", '
            '"_score": 1.3862944, '
            '"_source": {"Date": "2019-08-29T14:45:00.123Z"}'
            "}",
            "occurred": "2019-08-29T14:45:00Z",
            "dbotMirrorId": "123",
            "labels": [{"type": "Date", "value": "2019-08-29T14:45:00.123Z"}],
        },
        {
            "name": "Elasticsearch: Index: users, ID: 456",
            "rawJSON": "{"
            '"_index": "users", '
            '"_type": "_doc", '
            '"_id": "456", '
            '"_score": 0.9517491, '
            '"_source": {"Date": "2019-08-29T14:46:00.123456Z"}'
            "}",
            "occurred": "2019-08-29T14:46:00Z",
            "dbotMirrorId": "456",
            "labels": [{"type": "Date", "value": "2019-08-29T14:46:00.123456Z"}],
        },
    ]
)

MOCK_ES6_INCIDETNS_WITHOUT_LABELS = str(
    [
        {
            "name": "Elasticsearch: Index: users, ID: 123",
            "rawJSON": "{"
            '"_index": "users", '
            '"_type": "_doc", '
            '"_id": "123", '
            '"_score": 1.3862944, '
            '"_source": {"Date": "2019-08-29T14:45:00.123Z"}'
            "}",
            "occurred": "2019-08-29T14:45:00Z",
            "dbotMirrorId": "123",
        },
        {
            "name": "Elasticsearch: Index: users, ID: 456",
            "rawJSON": "{"
            '"_index": "users", '
            '"_type": "_doc", '
            '"_id": "456", '
            '"_score": 0.9517491, '
            '"_source": {"Date": "2019-08-29T14:46:00.123456Z"}'
            "}",
            "occurred": "2019-08-29T14:46:00Z",
            "dbotMirrorId": "456",
        },
    ]
)

ES_V7_RESPONSE_WITH_TIMESTAMP = {
    "took": 1,
    "timed_out": False,
    "_shards": {"total": 1, "successful": 1, "skipped": 0, "failed": 0},
    "hits": {
        "total": {"value": 9, "relation": "eq"},
        "max_score": 0.6814878,
        "hits": [
            {"_index": "customer", "_type": "doc", "_id": "123", "_score": 0.6814878, "_source": {"Date": "1572502634"}},
            {"_index": "customer", "_type": "doc", "_id": "456", "_score": 0.6814878, "_source": {"Date": "1572502640"}},
        ],
    },
}

MOCK_ES7_INCIDENTS_FROM_TIMESTAMP = str(
    [
        {
            "name": "Elasticsearch: Index: customer, ID: 123",
            "rawJSON": "{"
            '"_index": "customer", '
            '"_type": "doc", '
            '"_id": "123", '
            '"_score": 0.6814878, '
            '"_source": {"Date": "1572502634"}'
            "}",
            "occurred": "2019-10-31T06:17:14Z",
            "dbotMirrorId": "123",
            "labels": [{"type": "Date", "value": "1572502634"}],
        },
        {
            "name": "Elasticsearch: Index: customer, ID: 456",
            "rawJSON": "{"
            '"_index": "customer", '
            '"_type": "doc", '
            '"_id": "456", '
            '"_score": 0.6814878, '
            '"_source": {"Date": "1572502640"}'
            "}",
            "occurred": "2019-10-31T06:17:20Z",
            "dbotMirrorId": "456",
            "labels": [{"type": "Date", "value": "1572502640"}],
        },
    ]
)

MOCK_ES7_INCIDENTS_FROM_TIMESTAMP_WITHOUT_LABELS = str(
    [
        {
            "name": "Elasticsearch: Index: customer, ID: 123",
            "rawJSON": "{"
            '"_index": "customer", '
            '"_type": "doc", '
            '"_id": "123", '
            '"_score": 0.6814878, '
            '"_source": {"Date": "1572502634"}'
            "}",
            "occurred": "2019-10-31T06:17:14Z",
            "dbotMirrorId": "123",
        },
        {
            "name": "Elasticsearch: Index: customer, ID: 456",
            "rawJSON": "{"
            '"_index": "customer", '
            '"_type": "doc", '
            '"_id": "456", '
            '"_score": 0.6814878, '
            '"_source": {"Date": "1572502640"}'
            "}",
            "occurred": "2019-10-31T06:17:20Z",
            "dbotMirrorId": "456",
        },
    ]
)

MOCK_ES7_SCHEMA_INPUT = {"bytes": {"type": "long"}, "clientip": {"type": "ip"}}

MOCK_ES7_SCHEMA_OUTPUT = {"bytes": "type: long", "clientip": "type: ip"}

MOC_ES7_SERVER_RESPONSE = {
    "kibana_sample_data_logs": {
        "mappings": {
            "properties": {
                "@timestamp": {"type": "alias", "path": "timestamp"},
                "agent": {"type": "text", "fields": {"keyword": {"type": "keyword", "ignore_above": 256}}},
                "bytes": {"type": "long"},
                "clientip": {"type": "ip"},
                "event": {"properties": {"dataset": {"type": "keyword"}}},
                "extension": {"type": "text", "fields": {"keyword": {"type": "keyword", "ignore_above": 256}}},
                "geo": {
                    "properties": {
                        "coordinates": {"type": "geo_point"},
                        "dest": {"type": "keyword"},
                        "src": {"type": "keyword"},
                        "srcdest": {"type": "keyword"},
                    }
                },
                "host": {"type": "text", "fields": {"keyword": {"type": "keyword", "ignore_above": 256}}},
                "index": {"type": "text", "fields": {"keyword": {"type": "keyword", "ignore_above": 256}}},
                "ip": {"type": "ip"},
                "machine": {
                    "properties": {
                        "os": {"type": "text", "fields": {"keyword": {"type": "keyword", "ignore_above": 256}}},
                        "ram": {"type": "long"},
                    }
                },
                "memory": {"type": "double"},
                "message": {"type": "text", "fields": {"keyword": {"type": "keyword", "ignore_above": 256}}},
                "phpmemory": {"type": "long"},
                "referer": {"type": "keyword"},
                "request": {"type": "text", "fields": {"keyword": {"type": "keyword", "ignore_above": 256}}},
                "response": {"type": "text", "fields": {"keyword": {"type": "keyword", "ignore_above": 256}}},
                "tags": {"type": "text", "fields": {"keyword": {"type": "keyword", "ignore_above": 256}}},
                "timestamp": {"type": "date"},
                "url": {"type": "text", "fields": {"keyword": {"type": "keyword", "ignore_above": 256}}},
                "utc_time": {"type": "date"},
            }
        }
    }
}

MOCK_INDEX_RESPONSE = {
    "_index": "test-index",
    "_id": "1",
    "_version": 1,
    "result": "created",
    "_shards": {"total": 2, "successful": 2, "failed": 0},
    "_seq_no": 5,
    "_primary_term": 1,
}

MOCK_INDICES_STATISTICS_RESPONSE = {
    "index_name1": {"uuid": "1111", "health": "yellow", "status": "open", "total": {"docs": {"count": 2, "deleted": 0}}},
    "index_name2": {"uuid": "2222", "health": "green", "status": "closed", "total": {"docs": {"count": 40, "deleted": 2}}},
}

MOCK_PARAMS = [
    {
        "client_type": "Elasticsearch",
        "fetch_index": "customer",
        "fetch_time_field": "Date",
        "time_method": "Simple-Date",
        "map_labels": True,
        "credentials": {
            "identifier": "mock",
            "password": "demisto",
        },
    },
    {
        "client_type": "Elasticsearch",
        "fetch_index": "customer",
        "fetch_time_field": "Date",
        "time_method": "Simple-Date",
        "map_labels": False,
        "credentials": {
            "identifier": "mock",
            "password": "demisto",
        },
    },
    {
        "client_type": "OpenSearch",
        "fetch_index": "customer",
        "fetch_time_field": "Date",
        "time_method": "Simple-Date",
        "map_labels": True,
        "credentials": {
            "identifier": "mock",
            "password": "demisto",
        },
    },
]

PARAMS_V8 = {
    "client_type": "Elasticsearch_v8",
    "fetch_index": "customer",
    "fetch_time_field": "Date",
    "time_method": "Simple-Date",
    "map_labels": True,
    "credentials": {
        "identifier": "mock",
        "password": "demisto",
    },
}


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_context_creation_es7(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    from Elasticsearch_v2 import get_total_results, results_to_context

    base_page = 0
    size = 2
    total_dict, total_results = get_total_results(ES_V7_RESPONSE)
    query = "check"
    index = "customer"
    search_context, meta_headers, hit_tables, hit_headers = results_to_context(
        index, query, base_page, size, total_dict, ES_V7_RESPONSE
    )

    assert str(search_context) == MOCK_ES7_SEARCH_CONTEXT
    assert str(meta_headers) == "['Query', 'took', 'timed_out', 'total', 'max_score', 'Server', 'Page', 'Size', 'aggregations']"
    assert str(hit_tables) == MOCK_ES7_HIT_CONTEXT
    assert str(hit_headers) == "['_id', '_index', '_type', '_score', 'Date']"


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_context_creation_es6(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    from Elasticsearch_v2 import get_total_results, results_to_context

    base_page = 0
    size = 2
    total_dict, total_results = get_total_results(ES_V6_RESPONSE)
    query = "incident"
    index = "users"
    search_context, meta_headers, hit_tables, hit_headers = results_to_context(
        index, query, base_page, size, total_dict, ES_V6_RESPONSE
    )

    assert str(search_context) == MOCK_ES6_SEARCH_CONTEXT
    assert str(meta_headers) == "['Query', 'took', 'timed_out', 'total', 'max_score', 'Server', 'Page', 'Size', 'aggregations']"
    assert str(hit_tables) == MOCK_ES6_HIT_CONTEXT
    assert str(hit_headers) == "['_id', '_index', '_type', '_score', 'Date']"


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_incident_creation_e6(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    from Elasticsearch_v2 import results_to_incidents_datetime

    last_fetch = "2019-08-29T14:44:00Z"
    incidents, last_fetch2, _ = results_to_incidents_datetime(ES_V6_RESPONSE, last_fetch)

    # last fetch should not truncate the milliseconds
    assert str(last_fetch2) == "2019-08-29T14:46:00.123456+00:00"
    if params.get("map_labels"):
        assert str(incidents) == MOCK_ES6_INCIDETNS
    else:
        assert str(incidents) == MOCK_ES6_INCIDETNS_WITHOUT_LABELS


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_incident_creation_e7(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    from Elasticsearch_v2 import results_to_incidents_datetime

    last_fetch = "2019-08-27T17:59:00"
    incidents, last_fetch2, _ = results_to_incidents_datetime(ES_V7_RESPONSE, last_fetch)

    # last fetch should not truncate the milliseconds
    assert str(last_fetch2) == "2019-08-27T18:01:25.343212+00:00"
    if params.get("map_labels"):
        assert str(incidents) == MOCK_ES7_INCIDENTS
    else:
        assert str(incidents) == MOCK_ES7_INCIDENTS_WITHOUT_LABELS


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_timestamp_to_date_converter_seconds(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    mocker.patch("Elasticsearch_v2.TIME_METHOD", "Timestamp-Seconds")
    from Elasticsearch_v2 import timestamp_to_date

    seconds_since_epoch = "1572164838"
    assert str(timestamp_to_date(seconds_since_epoch)) == "2019-10-27 08:27:18"


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_timestamp_to_date_converter_milliseconds(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    mocker.patch("Elasticsearch_v2.TIME_METHOD", "Timestamp-Milliseconds")
    from Elasticsearch_v2 import timestamp_to_date

    milliseconds_since_epoch = "1572164838123"
    assert str(timestamp_to_date(milliseconds_since_epoch)) == "2019-10-27 08:27:18.123000"


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_incident_creation_with_timestamp_e7(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    mocker.patch("Elasticsearch_v2.TIME_METHOD", "Timestamp-Seconds")
    from Elasticsearch_v2 import results_to_incidents_timestamp

    lastfetch = int(datetime.strptime("2019-08-27T17:59:00Z", "%Y-%m-%dT%H:%M:%SZ").timestamp())
    incidents, last_fetch2, _ = results_to_incidents_timestamp(ES_V7_RESPONSE_WITH_TIMESTAMP, lastfetch)
    assert last_fetch2 == 1572502640
    if params.get("map_labels"):
        assert str(incidents) == MOCK_ES7_INCIDENTS_FROM_TIMESTAMP
    else:
        assert str(incidents) == MOCK_ES7_INCIDENTS_FROM_TIMESTAMP_WITHOUT_LABELS


# XSUP-72750: three documents sharing an identical sub-second timestamp at the fetch boundary.
ES_IDENTICAL_TIMESTAMP_RESPONSE = {
    "took": 1,
    "timed_out": False,
    "_shards": {"total": 1, "successful": 1, "skipped": 0, "failed": 0},
    "hits": {
        "total": {"value": 3, "relation": "eq"},
        "max_score": 1.0,
        "hits": [
            {"_index": "customer", "_type": "doc", "_id": "a1", "_source": {"Date": "2019-08-27T18:00:00.120000Z"}},
            {"_index": "customer", "_type": "doc", "_id": "a2", "_source": {"Date": "2019-08-27T18:00:00.120000Z"}},
            {"_index": "customer", "_type": "doc", "_id": "a3", "_source": {"Date": "2019-08-27T18:00:00.120000Z"}},
        ],
    },
}


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_datetime_identical_timestamps_no_data_loss_across_fetches(params, mocker):
    """
    XSUP-72750 regression test (Simple-Date / datetime path).

    Given:
        - Three documents that share an identical sub-second timestamp which is the
          fetch high-water-mark, split across two fetch cycles (first fetch returns
          only 'a1', second fetch returns all three because the query lower bound is now inclusive).
    When:
        - Running results_to_incidents_datetime twice, feeding the returned last_fetch and
          last_fetch_ids from the first call into the second (mimicking demisto.setLastRun/getLastRun).
    Then:
        - All three documents are ingested exactly once (no data loss, no duplicates).
    """
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)
    from Elasticsearch_v2 import results_to_incidents_datetime

    first_response = {"hits": {"hits": [ES_IDENTICAL_TIMESTAMP_RESPONSE["hits"]["hits"][0]]}}

    # First fetch: only 'a1' is available (page boundary cuts the rest off).
    incidents1, last_fetch1, ids1 = results_to_incidents_datetime(first_response, "2019-08-27T17:59:00Z")
    assert [inc["dbotMirrorId"] for inc in incidents1] == ["a1"]
    assert ids1 == ["a1"]

    # Second fetch: query is now gte the boundary timestamp, so all three are returned.
    incidents2, last_fetch2, ids2 = results_to_incidents_datetime(ES_IDENTICAL_TIMESTAMP_RESPONSE, last_fetch1, ids1)
    # 'a1' must be skipped (already ingested), 'a2' and 'a3' must be ingested.
    assert [inc["dbotMirrorId"] for inc in incidents2] == ["a2", "a3"]
    assert sorted(ids2) == ["a1", "a2", "a3"]

    # Overall: all three ingested exactly once.
    all_ingested = [inc["dbotMirrorId"] for inc in incidents1 + incidents2]
    assert sorted(all_ingested) == ["a1", "a2", "a3"]


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_datetime_identical_timestamps_no_duplicates_on_replay(params, mocker):
    """
    XSUP-72750 regression test (Simple-Date / datetime path) - idempotency.

    Given:
        - A fetch that already ingested all three identically-timestamped documents.
    When:
        - The exact same response is returned again (e.g. no new documents arrived).
    Then:
        - No incidents are produced (all skipped by _id de-duplication), so no duplicates.
    """
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)
    from Elasticsearch_v2 import results_to_incidents_datetime

    incidents1, last_fetch1, ids1 = results_to_incidents_datetime(ES_IDENTICAL_TIMESTAMP_RESPONSE, "2019-08-27T17:59:00Z")
    assert sorted(inc["dbotMirrorId"] for inc in incidents1) == ["a1", "a2", "a3"]

    # Replay the same response with the persisted state.
    incidents2, last_fetch2, ids2 = results_to_incidents_datetime(ES_IDENTICAL_TIMESTAMP_RESPONSE, last_fetch1, ids1)
    assert incidents2 == []
    assert sorted(ids2) == ["a1", "a2", "a3"]


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_timestamp_identical_timestamps_no_data_loss_across_fetches(params, mocker):
    """
    XSUP-72750 regression test (Timestamp path).

    Given:
        - Three documents that share an identical epoch-millisecond timestamp which is the
          fetch high-water-mark, split across two fetch cycles.
    When:
        - Running results_to_incidents_timestamp twice, feeding the returned last_fetch and
          last_fetch_ids from the first call into the second.
    Then:
        - All three documents are ingested exactly once (no data loss, no duplicates).
    """
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)
    mocker.patch("Elasticsearch_v2.TIME_METHOD", "Timestamp-Milliseconds")
    from Elasticsearch_v2 import results_to_incidents_timestamp

    ts = 1566928800120  # 2019-08-27T18:00:00.120Z in epoch milliseconds
    hit = {"_index": "customer", "_type": "doc", "_source": {"Date": ts}}
    response = {
        "hits": {
            "hits": [
                {**hit, "_id": "a1"},
                {**hit, "_id": "a2"},
                {**hit, "_id": "a3"},
            ]
        }
    }
    first_response = {"hits": {"hits": [{**hit, "_id": "a1"}]}}

    last_fetch = 1566928740000  # earlier than ts
    incidents1, last_fetch1, ids1 = results_to_incidents_timestamp(first_response, last_fetch)
    assert [inc["dbotMirrorId"] for inc in incidents1] == ["a1"]
    assert last_fetch1 == ts
    assert ids1 == ["a1"]

    incidents2, last_fetch2, ids2 = results_to_incidents_timestamp(response, last_fetch1, ids1)
    assert [inc["dbotMirrorId"] for inc in incidents2] == ["a2", "a3"]
    assert last_fetch2 == ts
    assert sorted(ids2) == ["a1", "a2", "a3"]


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_datetime_identical_timestamps_boundary_ids_persist_over_three_fetches(params, mocker):
    """
    XSUP-72750 regression test - continuity over three fetch cycles.

    Given:
        - Three documents sharing an identical boundary timestamp, revealed one per fetch.
    When:
        - Running results_to_incidents_datetime three times, always feeding back the persisted
          last_fetch and last_fetch_ids (mimicking demisto.getLastRun/setLastRun).
    Then:
        - Each document is ingested exactly once and the boundary id set never drifts,
          so no document is ever re-ingested as a duplicate on a later cycle.
    """
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)
    from Elasticsearch_v2 import results_to_incidents_datetime

    hits = ES_IDENTICAL_TIMESTAMP_RESPONSE["hits"]["hits"]
    fetch1 = {"hits": {"hits": hits[:1]}}
    fetch2 = {"hits": {"hits": hits[:2]}}
    fetch3 = {"hits": {"hits": hits[:3]}}

    incidents1, last_fetch, ids = results_to_incidents_datetime(fetch1, "2019-08-27T17:59:00Z")
    incidents2, last_fetch, ids = results_to_incidents_datetime(fetch2, last_fetch, ids)
    incidents3, last_fetch, ids = results_to_incidents_datetime(fetch3, last_fetch, ids)

    assert [inc["dbotMirrorId"] for inc in incidents1] == ["a1"]
    assert [inc["dbotMirrorId"] for inc in incidents2] == ["a2"]
    assert [inc["dbotMirrorId"] for inc in incidents3] == ["a3"]
    assert sorted(ids) == ["a1", "a2", "a3"]

    all_ingested = [inc["dbotMirrorId"] for inc in incidents1 + incidents2 + incidents3]
    assert sorted(all_ingested) == ["a1", "a2", "a3"]


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_datetime_newer_timestamp_resets_boundary_ids(params, mocker):
    """
    XSUP-72750 regression test - boundary id reset when a newer timestamp appears.

    Given:
        - A page containing some documents at the boundary timestamp and some at a strictly
          newer timestamp.
    When:
        - Running results_to_incidents_datetime with the boundary id already persisted.
    Then:
        - The previously-fetched boundary id is skipped, new documents are ingested, and the
          returned boundary id set contains only the ids at the new (newer) high-water-mark.
    """
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)
    from Elasticsearch_v2 import results_to_incidents_datetime

    response = {
        "hits": {
            "hits": [
                {"_index": "c", "_type": "doc", "_id": "a1", "_source": {"Date": "2019-08-27T18:00:00.120000Z"}},
                {"_index": "c", "_type": "doc", "_id": "a2", "_source": {"Date": "2019-08-27T18:00:00.120000Z"}},
                {"_index": "c", "_type": "doc", "_id": "b1", "_source": {"Date": "2019-08-27T18:05:00.500000Z"}},
            ]
        }
    }

    # 'a1' was already ingested at the previous boundary (18:00:00.120000).
    incidents, last_fetch, ids = results_to_incidents_datetime(response, "2019-08-27T18:00:00.120000Z", ["a1"])

    # 'a1' skipped, 'a2' and 'b1' ingested; the boundary advanced to b1's timestamp,
    # so only 'b1' remains as the boundary id.
    assert [inc["dbotMirrorId"] for inc in incidents] == ["a2", "b1"]
    assert ids == ["b1"]


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_format_to_iso(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    from Elasticsearch_v2 import format_to_iso

    date_string_1 = "2020-02-03T10:00:00"
    date_string_2 = "2020-02-03T10:00:00+02:00"
    date_string_3 = "2020-02-03T10:00:00-02:00"
    iso_format = "2020-02-03T10:00:00Z"
    assert format_to_iso(date_string_1) == iso_format
    assert format_to_iso(date_string_2) == iso_format
    assert format_to_iso(date_string_3) == iso_format
    assert format_to_iso(iso_format) == iso_format


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_elasticsearch_builder_called_with_username_password(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    from Elasticsearch_v2 import Elasticsearch, elasticsearch_builder

    es_mock = mocker.patch.object(Elasticsearch, "__init__", return_value=None)
    elasticsearch_builder(None)
    assert es_mock.call_args[1].get("http_auth") == ("mock", "demisto")
    assert es_mock.call_args[1].get("api_key") is None


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_elasticsearch_builder_called_with_no_creds(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    mocker.patch("Elasticsearch_v2.USERNAME", None)
    mocker.patch("Elasticsearch_v2.PASSWORD", None)
    from Elasticsearch_v2 import Elasticsearch, elasticsearch_builder

    es_mock = mocker.patch.object(Elasticsearch, "__init__", return_value=None)
    elasticsearch_builder(None)
    assert es_mock.call_args[1].get("http_auth") is None
    assert es_mock.call_args[1].get("api_key") is None
    assert es_mock.call_args[1].get("bearer_auth") is None


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_elasticsearch_builder_called_with_cred(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    mocker.patch("Elasticsearch_v2.USERNAME", "username")
    mocker.patch("Elasticsearch_v2.PASSWORD", "password")
    mocker.patch("Elasticsearch_v2.AUTH_TYPE", Elasticsearch_v2.BASIC_AUTH)
    from Elasticsearch_v2 import Elasticsearch, elasticsearch_builder

    es_mock = mocker.patch.object(Elasticsearch, "__init__", return_value=None)
    elasticsearch_builder(None)
    assert es_mock.call_args[1].get("http_auth")[0] == "username"
    assert es_mock.call_args[1].get("http_auth")[1] == "password"
    assert es_mock.call_args[1].get("api_key") is None
    assert es_mock.call_args[1].get("bearer_auth") is None

    mocker.patch("Elasticsearch_v2.AUTH_TYPE", Elasticsearch_v2.API_KEY_AUTH)
    mocker.patch("Elasticsearch_v2.API_KEY", "api_key_id")
    elasticsearch_builder(None)
    assert es_mock.call_args[1].get("http_auth") is None
    assert es_mock.call_args[1].get("api_key") == "api_key_id"
    assert es_mock.call_args[1].get("bearer_auth") is None

    mocker.patch("Elasticsearch_v2.AUTH_TYPE", Elasticsearch_v2.BEARER_AUTH)
    mocker.patch("Elasticsearch_v2.get_elastic_token", return_value="elastic_token")
    elasticsearch_builder(None)
    assert es_mock.call_args[1].get("http_auth") is None
    assert es_mock.call_args[1].get("api_key") is None
    assert es_mock.call_args[1].get("bearer_auth") == "elastic_token"


@pytest.mark.parametrize("params", MOCK_PARAMS)
def test_elasticsearch_parse_subtree(params, mocker):
    mocker.patch.object(demisto, "params", return_value=params)
    importlib.reload(Elasticsearch_v2)  # To reset the Elasticsearch client with the OpenSearch library
    from Elasticsearch_v2 import parse_subtree

    sub_tree = parse_subtree(MOCK_ES7_SCHEMA_INPUT)
    assert str(sub_tree) == str(MOCK_ES7_SCHEMA_OUTPUT)


# This is the class we want to test
"""
The get-mapping-fields command perform a GET /<index name>/_mapping http command
for e.g http://elasticserver.com/customers/_mapping the output is then formatted and arranged by the parse-tree function
The test created a mock response.
"""


class GetMapping:
    def fetch_json(self, url):
        response = requests.get(url)
        return response.json()


# This method will be used by the mock to replace requests.get
@patch("Elasticsearch_v2.FETCH_INDEX", "customer")
def mocked_requests_get(*args, **kwargs):
    class MockResponse:
        def __init__(self, json_data, status_code):
            self.json_data = json_data
            self.status_code = status_code

        def json(self):
            return self.json_data

    if args[0] == "http://someurl.com/" + "index" + "/_mapping":
        return MockResponse(MOC_ES7_SERVER_RESPONSE, 200)
    else:
        return MockResponse(None, 404)


# Our test case class
class GetMappingFields(unittest.TestCase):
    # We patch 'requests.get' with our own method. The mock object is passed in to our test case method.
    @mock.patch("requests.get", side_effect=mocked_requests_get)
    def test_fetch(self, mock_get):
        # Assert requests.get calls
        gmf = GetMapping()
        server_response = gmf.fetch_json("http://someurl.com/" + "index" + "/_mapping")
        assert server_response == MOC_ES7_SERVER_RESPONSE


class TestIncidentLabelMaker(unittest.TestCase):
    def test_sanity(self):
        from Elasticsearch_v2 import incident_label_maker

        sources = {
            "first_name": "John",
            "sur_name": "Snow",
        }
        expected_labels = [
            {"type": "first_name", "value": "John"},
            {"type": "sur_name", "value": "Snow"},
        ]

        labels = incident_label_maker(sources)
        assert labels == expected_labels

    def test_complex_value(self):
        from Elasticsearch_v2 import incident_label_maker

        sources = {
            "name": "Ash",
            "action": "catch",
            "targets": ["Pikachu", "Charmander", "Squirtle", "Bulbasaur"],
        }
        expected_labels = [
            {
                "type": "name",
                "value": "Ash",
            },
            {
                "type": "action",
                "value": "catch",
            },
            {
                "type": "targets",
                "value": '["Pikachu", "Charmander", "Squirtle", "Bulbasaur"]',
            },
        ]

        labels = incident_label_maker(sources)
        assert labels == expected_labels


@pytest.mark.parametrize(
    "time_method, last_fetch, time_range_start, time_range_end, result",
    [
        (
            "Timestamp-Milliseconds",
            "",
            "1.1.2000 12:00:00Z",
            "2.1.2000 12:00:00Z",
            {"range": {"time_field": {"gte": 946728000000, "lt": 949406400000}}},
        ),
        (
            "Timestamp-Milliseconds",
            946728000000,
            "",
            "2.1.2000 12:00:00Z",
            {"range": {"time_field": {"gte": 946728000000, "lt": 949406400000}}},
        ),
        ("Timestamp-Milliseconds", "", "", "2.1.2000 12:00:00Z", {"range": {"time_field": {"lt": 949406400000}}}),
        (
            "Simple-Date",
            "2.1.2000 12:00:00.000000",
            "",
            "",
            {"range": {"time_field": {"gte": "2.1.2000 12:00:00.000000", "format": Elasticsearch_v2.ES_DEFAULT_DATETIME_FORMAT}}},
        ),
    ],
)
def test_get_time_range(time_method, last_fetch, time_range_start, time_range_end, result):
    Elasticsearch_v2.TIME_METHOD = time_method
    from Elasticsearch_v2 import get_time_range

    assert get_time_range(last_fetch, time_range_start, time_range_end, "time_field") == result


@pytest.mark.parametrize(
    "time_method, time_range_start, expected_time_zone",
    [
        ("Simple-Date", "2024-01-15T10:30:00+02:00", "+02:00"),
        ("Simple-Date", "2024-01-15T10:30:00-05:00", "-05:00"),
        ("Simple-Date", "2024-01-15T10:30:00+03:30", "+03:30"),
        ("Simple-Date", "2024-01-15T10:30:00-11:00", "-11:00"),
        ("Timestamp-Seconds", "2024-01-15T10:30:00+02:00", "+02:00"),
        ("Timestamp-Milliseconds", "2024-01-15T10:30:00-05:00", "-05:00"),
        ("Simple-Date", "2024-01-15T10:30:00Z", None),
        ("Simple-Date", "2024-01-15T10:30:00", None),
    ],
)
def test_get_time_range_with_utc_offset(time_method, time_range_start, expected_time_zone):
    """
    Test that UTC offset is correctly extracted from time_range_start and added to range_dict.

    Given:
        - A time_range_start with various UTC offset formats (+HH:MM or -HH:MM)
        - Different time methods (Simple-Date, Timestamp-Seconds, Timestamp-Milliseconds)

    When:
        - Calling get_time_range with the time_range_start parameter

    Then:
        - The UTC offset should be extracted and added to range_dict as 'time_zone'
        - If no UTC offset is present (Z or no offset), time_zone should not be in range_dict
    """
    Elasticsearch_v2.TIME_METHOD = time_method
    from Elasticsearch_v2 import get_time_range

    result = get_time_range(last_fetch=None, time_range_start=time_range_start, time_range_end=None, time_field="time_field")

    if expected_time_zone:
        assert "time_zone" in result["range"]["time_field"]
        assert result["range"]["time_field"]["time_zone"] == expected_time_zone
    else:
        assert "time_zone" not in result["range"]["time_field"]


def test_build_eql_body():
    from Elasticsearch_v2 import build_eql_body

    assert build_eql_body(None, None, None, None, None, None, None) == {}
    assert build_eql_body("query", "fields", "size", "tiebreaker_field", "timestamp_field", "event_category_field", "filter") == {
        "query": "query",
        "fields": "fields",
        "size": "size",
        "tiebreaker_field": "tiebreaker_field",
        "timestamp_field": "timestamp_field",
        "event_category_field": "event_category_field",
        "filter": "filter",
    }


first_case_all_with_empty_string = (
    "",
    {"a": {"mappings": {"properties": {"example": {}}}}},
    {"a": {"_id": "doc_id", "_index": "a", "_source": {"example": "type: "}}},
)
second_case_with_prefix_and_wildcard = (
    ".internal.alerts-*",
    {
        ".internal.alerts-security": {"mappings": {"properties": {"example": {}}}},
        ".internal": {"mappings": {"properties": {"example": {}}}},
    },
    {".internal.alerts-security": {"_id": "doc_id", "_index": ".internal.alerts-security", "_source": {"example": "type: "}}},
)
third_regular_case = (
    "a",
    {"a": {"mappings": {"properties": {"example": {}}}}},
    {"a": {"_id": "doc_id", "_index": "a", "_source": {"example": "type: "}}},
)


@pytest.mark.parametrize(
    "indexes, response, expected_result",
    [first_case_all_with_empty_string, second_case_with_prefix_and_wildcard, third_regular_case],
)
def test_get_mapping_fields_command(mocker, indexes, response, expected_result):
    class ResponseMockObject:
        def json(self):
            return response

    mocker.patch("Elasticsearch_v2.FETCH_INDEX", indexes)
    mocker.patch("Elasticsearch_v2.requests.get", return_value=ResponseMockObject())
    result = Elasticsearch_v2.get_mapping_fields_command()
    assert result == expected_result


def test_search_command_with_query_dsl(mocker):
    """
    Given
      - index to the search command with query_dsl

    When
    - executing the search command

    Then
     - make sure that the index is being taken from the command arguments and not from integration parameters
     - make sure that the size / page arguments are getting applied when using query_dsl
    """
    import Elasticsearch_v2

    Elasticsearch_v2.FETCH_INDEX = "index from parameter"
    index_from_arg = "index from arg"
    mocker.patch.object(
        demisto,
        "args",
        return_value={"index": index_from_arg, "query_dsl": '{"query": {"match": {"name": "test"}}}', "size": "5", "page": "0"},
    )
    search_mock = mocker.patch.object(Elasticsearch_v2.Elasticsearch, "search", return_value=ES_V7_RESPONSE)
    mocker.patch.object(Elasticsearch_v2.Elasticsearch, "__init__", return_value=None)
    Elasticsearch_v2.search_command({})
    assert search_mock.call_args.kwargs["index"] == [index_from_arg]
    assert search_mock.call_args.kwargs["body"] == {"query": {"match": {"name": "test"}}, "size": 5, "from": 0}


@pytest.mark.parametrize(
    "raw_query_body",
    [
        ({"query": {"match": {"name": "test"}}, "size": 2, "from": 1}),
        ({"query": {"match": {"name": "test"}}, "size": 2}),
        ({"query": {"match": {"name": "test"}}, "from": 3}),
        ({"query": {"match": {"name": "test"}}}),
    ],
)
def test_execute_raw_query(mocker, raw_query_body):
    """
    Given
      - index and elastic search objects

    When
    - executing execute_raw_query function with two response: first an exception and second a correct response.

    Then
     - make sure that no exception was raised from the function.
     - make sure the response came back correctly.
    """
    import Elasticsearch_v2

    Elasticsearch_v2.FETCH_INDEX = "index from parameter"
    mocker.patch.object(Elasticsearch_v2.Elasticsearch, "search", return_value=ES_V7_RESPONSE)
    mocker.patch.object(Elasticsearch_v2.Elasticsearch, "__init__", return_value=None)
    es = Elasticsearch_v2.elasticsearch_builder({})
    assert Elasticsearch_v2.execute_raw_query(es, json.dumps(raw_query_body)) == ES_V7_RESPONSE


@patch.dict("os.environ", {"DEMISTO_PARAMS": str(PARAMS_V8)})
@pytest.mark.parametrize(
    "raw_query_body",
    [
        ({"query": {"match": {"name": "test"}}, "size": 2, "from": 1}),
        ({"query": {"match": {"name": "test"}}, "size": 2}),
        ({"query": {"match": {"name": "test"}}, "from": 3}),
        ({"query": {"match": {"name": "test"}}}),
    ],
)
def test_execute_raw_query_v8(mocker, raw_query_body):
    """
    Given
      - index and elastic search objects
      - instance configured to v8

    When
    - executing execute_raw_query function with query_dsl body

    Then
     - make sure that no exception was raised from the function.
     - make sure the response came back correctly.
     - make sure the query body can be serialized an does not throw errors.
    """
    import Elasticsearch_v2
    from elastic_transport import RequestsHttpNode

    Elasticsearch_v2.RequestsHttpNode = RequestsHttpNode

    class CustomExecute:
        def to_dict():  # type: ignore
            return ES_V8_RESPONSE

    mocker.patch.object(Elasticsearch_v2, "ELASTIC_SEARCH_CLIENT", Elasticsearch_v2.ELASTICSEARCH_V8)
    mocker.patch.object(Elasticsearch_v2.Elasticsearch, "search", return_value=ES_V7_RESPONSE)
    mocker.patch.object(Elasticsearch_v2.Search, "execute", return_value=CustomExecute)
    mocker.patch.object(Elasticsearch_v2.Elasticsearch, "__init__", return_value=None)
    mocker.patch.object(RequestsHttpNode, "__init__", return_value=None)

    es = Elasticsearch_v2.elasticsearch_builder({})
    assert Elasticsearch_v2.execute_raw_query(es, json.dumps(raw_query_body)) == ES_V8_RESPONSE


@pytest.mark.parametrize(
    "date_time, time_method, expected_time",
    [
        ("123456", "Timestamp-Seconds", 123456),
        ("123456", "Timestamp-Milliseconds", 123456),
        (dateparser.parse("July 1, 2023"), "Simple-Date", "2023-07-01 00:00:00.000000"),
        (dateparser.parse("2023-07-01 23:24:25.123456"), "Simple-Date", "2023-07-01 23:24:25.123456"),
    ],
)
def test_convert_date_to_timestamp(date_time, time_method, expected_time):
    """
    Given
      - A python datetime object.
      - The time_method parameter ('Timestamp-Seconds', 'Timestamp-Milliseconds', 'Simple-Date').

    When
        - Executing convert_date_to_timestamp function.

    Then
        - Make sure that the returned datetime is as expected with the correct format.
    """
    Elasticsearch_v2.TIME_METHOD = time_method
    assert Elasticsearch_v2.convert_date_to_timestamp(date_time) == expected_time


def test_index_document(mocker):
    """
    Given
      - index name, document in JSON format, id of document

    When
    - executing index_document function.

    Then
     - Make sure that the returned function response is as expected with the correct format
    """
    import Elasticsearch_v2

    mocker.patch.object(Elasticsearch_v2.Elasticsearch, "index", return_value=MOCK_INDEX_RESPONSE)
    mocker.patch.object(Elasticsearch_v2.Elasticsearch, "__init__", return_value=None)
    assert Elasticsearch_v2.index_document({"index_name": "test-index", "document": "{}", "id": "1"}, "") == MOCK_INDEX_RESPONSE


def test_index_document_command(mocker):
    """
    Given
      - index name, document in JSON format, id of document

    When
    - executing index_document_command function.

    Then
     - Make sure that the returned function response is as expected with the correct format
    """
    import Elasticsearch_v2

    mocker.patch.object(Elasticsearch_v2.Elasticsearch, "index", return_value=MOCK_INDEX_RESPONSE)
    mocker.patch.object(Elasticsearch_v2.Elasticsearch, "__init__", return_value=None)
    command_result = Elasticsearch_v2.index_document_command({"index_name": "test-index", "document": "{}", "id": "1"}, "")
    expected_index_context = {
        "id": MOCK_INDEX_RESPONSE.get("_id", ""),
        "index": MOCK_INDEX_RESPONSE.get("_index", ""),
        "version": MOCK_INDEX_RESPONSE.get("_version", ""),
        "result": MOCK_INDEX_RESPONSE.get("result", ""),
    }
    expected_human_readable = (
        "### Indexed document\n|ID|Index name|Version|Result|\n|---|---|---|---|\n| 1 | test-index | 1 | created |\n"
    )

    assert command_result.outputs == expected_index_context
    assert command_result.readable_output == expected_human_readable
    assert command_result.outputs_prefix == "Elasticsearch.Index"
    assert command_result.raw_response == MOCK_INDEX_RESPONSE
    assert command_result.outputs_key_field == "id"


def test_get_value_by_dot_notation():
    """
    GIVEN a dictionary and a key in dot notation
    WHEN get_value_by_dot_notation is called
    THEN it should return the value corresponding to the key
    """
    dictionary = {"a": {"b": {"c": 123}}, "x": {"y": 456}}
    key = "a.b.c"

    result = Elasticsearch_v2.get_value_by_dot_notation(dictionary, key)

    assert result == 123


def test_key_not_found():
    """
    GIVEN a dictionary and a key in dot notation that does not exist
    WHEN get_value_by_dot_notation is called
    THEN it should return None
    """
    dictionary = {"a": {"b": True}, "x": {"y": 456}}
    key = "a.b.d"  # Key 'a.b.d' does not exist

    result = Elasticsearch_v2.get_value_by_dot_notation(dictionary, key)

    assert result is None


@pytest.mark.parametrize(
    "limit, all_results, expected_context",
    [
        (
            1,
            True,
            [
                {
                    "Name": "index_name1",
                    "Status": "open",
                    "Health": "yellow",
                    "UUID": "1111",
                    "Documents Count": 2,
                    "Documents Deleted": 0,
                },
                {
                    "Name": "index_name2",
                    "Status": "closed",
                    "Health": "green",
                    "UUID": "2222",
                    "Documents Count": 40,
                    "Documents Deleted": 2,
                },
            ],
        ),
        (
            1,
            False,
            [
                {
                    "Name": "index_name1",
                    "Status": "open",
                    "Health": "yellow",
                    "UUID": "1111",
                    "Documents Count": 2,
                    "Documents Deleted": 0,
                }
            ],
        ),
    ],
    ids=[
        "Test get indices statistics with a limit and all_results=True",
        "Test get indices statistics with a limit and all_results=False",
    ],
)
def test_get_indices_statistics_command(mocker, limit, all_results, expected_context):
    """
    Tests the 'get_indices_statistics' integration command.
    Given
      1. Elastic search client, a limit of 1, all_results arg set to True.
      2. Elastic search client, a limit of 1, all_results arg set to False.

    When
    - Running the get_indices_statistics_command function.

    Then
     - Make sure that the returned function response includes the indices mocked data as expected:
        1. All results were returned (2 indices).
        2. Only the first index's data was returned.
    """
    import Elasticsearch_v2

    mocker.patch.object(Elasticsearch_v2, "get_indices_statistics", return_value=MOCK_INDICES_STATISTICS_RESPONSE)
    mocker.patch.object(Elasticsearch_v2.Elasticsearch, "__init__", return_value=None)
    command_result = Elasticsearch_v2.get_indices_statistics_command({"limit": limit, "all_results": all_results}, "")

    assert command_result.outputs == expected_context
    assert "Indices Statistics:" in command_result.readable_output
    assert command_result.outputs_prefix == "Elasticsearch.IndexStatistics"
    assert command_result.raw_response == MOCK_INDICES_STATISTICS_RESPONSE
    assert command_result.outputs_key_field == "UUID"


@pytest.mark.parametrize(
    "server_details, server_version, client_version",
    [
        (
            {
                "name": "test1",
                "cluster_name": "elasticsearch",
                "cluster_uuid": "test_id",
                "version": {
                    "number": "7.3.0",
                },
            },
            "7.3.0",
            "Elasticsearch_v8",
        ),
        (
            {
                "name": "test2",
                "cluster_name": "elasticsearch",
                "cluster_uuid": "test_id",
                "version": {
                    "number": "8.4.1",
                },
            },
            "8.4.1",
            "Elasticsearch",
        ),
    ],
    ids=[
        "Test miss configuration error - server version is 7 while client version is 8",
        "Test miss configuration error - server version is 8 while client version is 7",
    ],
)
def test_verify_es_server_version_errors(mocker, server_details, server_version, client_version):
    """
    Tests the 'verify_es_server_version' function's logic.

    Given
      1. Elastic search server details (response json of the requests.get) - server version is 7.3.0.
         Integration parameter - client type - is set to 'Elasticsearch_v8.
      2. Elastic search server details (response json of the requests.get) - server version is 8.4.1.
         Integration parameter - client type - is set to 'Elasticsearch. (v7 and below)

    When
    - Running the verify_es_server_version function.

    Then
     - Make sure that the expected error message is raised.
    """
    import Elasticsearch_v2

    mocker.patch("Elasticsearch_v2.ELASTIC_SEARCH_CLIENT", new=client_version)
    with pytest.raises(ValueError) as e:
        Elasticsearch_v2.verify_es_server_version(server_details)
    assert server_version in str(e.value)


def test_search_command_with_query_esql(mocker):
    """
    Given
      - query to the search command with esql

    When
    - executing the es-esql-search command

    Then
     - Make sure that the expected message is returned.
    """
    MOCKER_RES = {"columns": [{"name": "col_1"}, {"name": "col_2"}], "values": [["val_1", "val_2"], ["val_1_1", "val_2_2"]]}
    EXPECTED_HEADERS = {
        "Content-Type": "application/vnd.elasticsearch+json; compatible-with=9",
        "Accept": "application/vnd.elasticsearch+json; compatible-with=9",
    }
    magic_mock = MagicMock()
    magic_mock.body = MOCKER_RES
    magic_mock.__getitem__.side_effect = lambda key: magic_mock.body[key]

    import Elasticsearch_v2

    mocked_builder = mocker.patch.object(Elasticsearch_v2, "elasticsearch_builder")
    mocked_builder().perform_request.return_value = magic_mock
    mocker.patch.object(Elasticsearch_v2, "ELASTIC_SEARCH_CLIENT", new=Elasticsearch_v2.ELASTICSEARCH_V9)

    query = """FROM alerts | WHERE alertDetails.alertuser LIKE "*karl*"| KEEP *"""
    res = Elasticsearch_v2.search_esql_command({"query": query, "limit": 2}, {})

    call_args = mocked_builder().perform_request.call_args[1]
    assert res.outputs_prefix == "Elasticsearch.ESQLSearch"
    assert res.outputs == [{"col_1": "val_1", "col_2": "val_2"}, {"col_1": "val_1_1", "col_2": "val_2_2"}]
    assert res.raw_response == magic_mock.body
    assert call_args["headers"] == EXPECTED_HEADERS
    assert call_args["body"] == {"query": f"{query}| LIMIT 2"}


class TestGetElasticToken:
    """Tests for the get_elastic_token function."""

    @pytest.fixture
    def mock_integration_context(self, mocker):
        """Fixture to mock integration context functions."""
        mock_get = mocker.patch("Elasticsearch_v2.get_integration_context")
        mock_set = mocker.patch("Elasticsearch_v2.set_integration_context")
        return mock_get, mock_set

    @pytest.fixture
    def mock_requests_post(self, mocker):
        """Fixture to mock requests.post."""
        return mocker.patch("Elasticsearch_v2.requests.post")

    def test_get_elastic_token_existing_valid_token(self, mocker, mock_integration_context):
        """
        Given:
            - An existing valid access token in integration context that hasn't expired
        When:
            - Calling get_elastic_token
        Then:
            - Return the existing access token without making any API calls
        """
        import Elasticsearch_v2

        mock_get, mock_set = mock_integration_context
        future_time = (datetime.now() + timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%SZ")

        mock_get.return_value = {
            "access_token": "existing_valid_token",
            "access_token_expires_in": future_time,
            "refresh_token": "refresh_token",
            "refresh_token_expires_in": future_time,
        }

        mocker.patch("Elasticsearch_v2.USERNAME", "test_user")
        mocker.patch("Elasticsearch_v2.PASSWORD", "test_pass")
        mocker.patch("Elasticsearch_v2.SERVER", "http://test-server")
        mocker.patch("Elasticsearch_v2.INSECURE", True)

        result = Elasticsearch_v2.get_elastic_token()

        assert result == "existing_valid_token"

    def test_get_elastic_token_expired_token_valid_refresh(self, mocker, mock_integration_context, mock_requests_post):
        """
        Given:
            - An expired access token but a valid refresh token in integration context
        When:
            - Calling get_elastic_token
        Then:
            - Use the refresh token to get a new access token
            - Update the integration context with new tokens
            - Return the new access token
        """
        import Elasticsearch_v2

        mock_get, mock_set = mock_integration_context
        past_time = (datetime.now() - timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%SZ")
        future_time = (datetime.now() + timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%SZ")

        mock_get.return_value = {
            "access_token": "expired_token",
            "access_token_expires_in": past_time,
            "refresh_token": "valid_refresh_token",
            "refresh_token_expires_in": future_time,
        }

        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = {
            "access_token": "new_access_token",
            "refresh_token": "new_refresh_token",
            "expires_in": 3600,
        }
        mock_requests_post.return_value = mock_response

        mocker.patch("Elasticsearch_v2.USERNAME", "test_user")
        mocker.patch("Elasticsearch_v2.PASSWORD", "test_pass")
        mocker.patch("Elasticsearch_v2.SERVER", "http://test-server")
        mocker.patch("Elasticsearch_v2.INSECURE", True)

        result = Elasticsearch_v2.get_elastic_token()

        assert result == "new_access_token"
        assert mock_requests_post.call_count == 1
        call_args = mock_requests_post.call_args
        assert call_args[1]["json"]["grant_type"] == "refresh_token"
        assert call_args[1]["json"]["refresh_token"] == "valid_refresh_token"

    def test_get_elastic_token_password_grant(self, mocker, mock_integration_context, mock_requests_post):
        """
        Given:
            - No existing tokens or expired refresh token
        When:
            - Calling get_elastic_token
        Then:
            - Perform password grant authentication
            - Store new tokens in integration context
            - Return the new access token
        """
        import Elasticsearch_v2

        mock_get, mock_set = mock_integration_context
        mock_get.return_value = {}

        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = {
            "access_token": "password_grant_token",
            "refresh_token": "password_grant_refresh",
            "expires_in": 3600,
        }
        mock_requests_post.return_value = mock_response

        mocker.patch("Elasticsearch_v2.USERNAME", "test_user")
        mocker.patch("Elasticsearch_v2.PASSWORD", "test_pass")
        mocker.patch("Elasticsearch_v2.SERVER", "http://test-server")
        mocker.patch("Elasticsearch_v2.INSECURE", True)

        result = Elasticsearch_v2.get_elastic_token()

        assert result == "password_grant_token"
        assert mock_requests_post.call_count == 1
        call_args = mock_requests_post.call_args
        assert call_args[1]["json"]["grant_type"] == "password"
        assert call_args[1]["json"]["username"] == "test_user"
        assert call_args[1]["json"]["password"] == "test_pass"

    def test_get_elastic_token_missing_credentials(self, mocker, mock_integration_context):
        """
        Given:
            - Missing username or password
        When:
            - Calling get_elastic_token
        Then:
            - Raise DemistoException with appropriate error message
        """
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        mock_get, mock_set = mock_integration_context
        mock_get.return_value = {}

        mocker.patch("Elasticsearch_v2.USERNAME", None)
        mocker.patch("Elasticsearch_v2.PASSWORD", "test_pass")
        mocker.patch("Elasticsearch_v2.SERVER", "http://test-server")
        mocker.patch("Elasticsearch_v2.INSECURE", True)

        with pytest.raises(DemistoException) as exc_info:
            Elasticsearch_v2.get_elastic_token()

        assert "username or password fields are missing" in str(exc_info.value)

    def test_get_elastic_token_refresh_fails_fallback_to_password(self, mocker, mock_integration_context, mock_requests_post):
        """
        Given:
            - Expired access token and valid refresh token
            - Refresh token request fails
        When:
            - Calling get_elastic_token
        Then:
            - Attempt refresh token flow first
            - Fall back to password grant when refresh fails
            - Return new access token from password grant
        """
        import Elasticsearch_v2

        mock_get, mock_set = mock_integration_context
        past_time = (datetime.now() - timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%SZ")
        future_time = (datetime.now() + timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%SZ")

        mock_get.return_value = {
            "access_token": "expired_token",
            "access_token_expires_in": past_time,
            "refresh_token": "valid_refresh_token",
            "refresh_token_expires_in": future_time,
        }

        # First call (refresh) fails, second call (password grant) succeeds
        mock_refresh_response = MagicMock()
        mock_refresh_response.status_code = 401

        mock_password_response = MagicMock()
        mock_password_response.status_code = 200
        mock_password_response.json.return_value = {
            "access_token": "new_password_token",
            "refresh_token": "new_refresh_token",
            "expires_in": 3600,
        }

        mock_requests_post.side_effect = [mock_refresh_response, mock_password_response]

        mocker.patch("Elasticsearch_v2.USERNAME", "test_user")
        mocker.patch("Elasticsearch_v2.PASSWORD", "test_pass")
        mocker.patch("Elasticsearch_v2.SERVER", "http://test-server")
        mocker.patch("Elasticsearch_v2.INSECURE", True)

        result = Elasticsearch_v2.get_elastic_token()

        assert result == "new_password_token"
        assert mock_requests_post.call_count == 2
        # Verify first call was refresh token
        assert mock_requests_post.call_args_list[0][1]["json"]["grant_type"] == "refresh_token"
        # Verify second call was password grant
        assert mock_requests_post.call_args_list[1][1]["json"]["grant_type"] == "password"

    def test_get_elastic_token_authentication_failure(self, mocker, mock_integration_context, mock_requests_post):
        """
        Given:
            - No existing tokens
            - Password grant authentication fails
        When:
            - Calling get_elastic_token
        Then:
            - Raise DemistoException with authentication failure message
        """
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        mock_get, mock_set = mock_integration_context
        mock_get.return_value = {}

        mock_response = MagicMock()
        mock_response.status_code = 401
        reason = "unable to authenticate user [test_user] for REST request [/_security/oauth2/token]"
        mock_response.text = json.dumps({"error": {"reason": reason}})
        mock_requests_post.return_value = mock_response

        mocker.patch("Elasticsearch_v2.USERNAME", "test_user")
        mocker.patch("Elasticsearch_v2.PASSWORD", "wrong_pass")
        mocker.patch("Elasticsearch_v2.SERVER", "http://test-server")
        mocker.patch("Elasticsearch_v2.INSECURE", True)

        with pytest.raises(DemistoException) as exc_info:
            Elasticsearch_v2.get_elastic_token()

        assert reason in str(exc_info.value)


class TestGetKibanaBaseUrl:
    """Tests for the get_kibana_base_url function."""

    def test_get_kibana_base_url_success(self, mocker):
        """
        Given:
            - No configured Kibana Server URL, and a Server URL containing the ".es." Elastic Cloud subdomain segment
        When:
            - Calling get_kibana_base_url
        Then:
            - Return the URL with ".es." replaced by ".kb."
        """
        mocker.patch.dict(Elasticsearch_v2.PARAMS, {"kibana_url": ""})
        mocker.patch("Elasticsearch_v2.SERVER", "https://my-deployment-af38b6.es.us-central1.gcp.cloud.es.io")

        result = Elasticsearch_v2.get_kibana_base_url()

        assert result == "https://my-deployment-af38b6.kb.us-central1.gcp.cloud.es.io"

    def test_get_kibana_base_url_from_configured_param(self, mocker):
        """
        Given:
            - A configured Kibana Server URL of an on-premises deployment, and a Server URL without ".es."
        When:
            - Calling get_kibana_base_url
        Then:
            - Return the configured Kibana Server URL as-is, without attempting any derivation
        """
        mocker.patch.dict(Elasticsearch_v2.PARAMS, {"kibana_url": "https://kibana.example.com:5601"})
        mocker.patch("Elasticsearch_v2.SERVER", "https://on-prem-elastic.example.com:9200")

        result = Elasticsearch_v2.get_kibana_base_url()

        assert result == "https://kibana.example.com:5601"

    def test_get_kibana_base_url_configured_param_strips_trailing_slash(self, mocker):
        """
        Given:
            - A configured Kibana Server URL that ends with a trailing slash
        When:
            - Calling get_kibana_base_url
        Then:
            - Return the URL without the trailing slash, so paths are joined correctly
        """
        mocker.patch.dict(Elasticsearch_v2.PARAMS, {"kibana_url": "https://kibana.example.com:5601/"})
        mocker.patch("Elasticsearch_v2.SERVER", "https://on-prem-elastic.example.com:9200")

        result = Elasticsearch_v2.get_kibana_base_url()

        assert result == "https://kibana.example.com:5601"

    def test_get_kibana_base_url_configured_param_takes_precedence(self, mocker):
        """
        Given:
            - Both a configured Kibana Server URL and an Elastic Cloud Server URL containing ".es."
        When:
            - Calling get_kibana_base_url
        Then:
            - Return the configured Kibana Server URL, which takes precedence over the derivation
        """
        mocker.patch.dict(Elasticsearch_v2.PARAMS, {"kibana_url": "https://custom-kibana.example.com"})
        mocker.patch("Elasticsearch_v2.SERVER", "https://my-deployment-af38b6.es.us-central1.gcp.cloud.es.io")

        result = Elasticsearch_v2.get_kibana_base_url()

        assert result == "https://custom-kibana.example.com"

    def test_get_kibana_base_url_missing_es_segment(self, mocker):
        """
        Given:
            - No configured Kibana Server URL, and a Server URL that does not contain the ".es." segment
        When:
            - Calling get_kibana_base_url
        Then:
            - Raise a DemistoException instructing the user to configure the Kibana Server URL
        """
        from CommonServerPython import DemistoException

        mocker.patch.dict(Elasticsearch_v2.PARAMS, {"kibana_url": ""})
        mocker.patch("Elasticsearch_v2.SERVER", "https://on-prem-elastic.example.com:9200")

        with pytest.raises(DemistoException) as exc_info:
            Elasticsearch_v2.get_kibana_base_url()

        assert "Could not determine the Kibana URL" in str(exc_info.value)
        assert "Kibana Server URL" in str(exc_info.value)


class TestGetKibanaAuthHeaders:
    """Tests for the get_kibana_auth_headers function."""

    def test_api_key_auth(self, mocker):
        """
        Given:
            - Auth type is API key auth
        When:
            - Calling get_kibana_auth_headers
        Then:
            - Return an Authorization header built from the API key
        """
        import Elasticsearch_v2

        mocker.patch("Elasticsearch_v2.AUTH_TYPE", Elasticsearch_v2.API_KEY_AUTH)
        mocker.patch("Elasticsearch_v2.API_KEY", ("id123", "secret456"))

        headers = Elasticsearch_v2.get_kibana_auth_headers()

        assert headers["Authorization"].startswith("ApiKey ")

    def test_bearer_auth(self, mocker):
        """
        Given:
            - Auth type is Bearer auth
        When:
            - Calling get_kibana_auth_headers
        Then:
            - Return an Authorization header built from the elastic OAuth token
        """
        import Elasticsearch_v2

        mocker.patch("Elasticsearch_v2.AUTH_TYPE", Elasticsearch_v2.BEARER_AUTH)
        mocker.patch("Elasticsearch_v2.get_elastic_token", return_value="my-token")

        headers = Elasticsearch_v2.get_kibana_auth_headers()

        assert headers["Authorization"] == "Bearer my-token"

    def test_basic_auth(self, mocker):
        """
        Given:
            - Auth type is Basic auth with username and password configured
        When:
            - Calling get_kibana_auth_headers
        Then:
            - Return a Basic Authorization header
        """
        import Elasticsearch_v2

        mocker.patch("Elasticsearch_v2.AUTH_TYPE", Elasticsearch_v2.BASIC_AUTH)
        mocker.patch("Elasticsearch_v2.USERNAME", "user")
        mocker.patch("Elasticsearch_v2.PASSWORD", "pass")

        headers = Elasticsearch_v2.get_kibana_auth_headers()

        assert headers["Authorization"].startswith("Basic ")

    def test_missing_credentials_raises(self, mocker):
        """
        Given:
            - Auth type is Basic auth but username/password are missing
        When:
            - Calling get_kibana_auth_headers
        Then:
            - Raise a DemistoException
        """
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        mocker.patch("Elasticsearch_v2.AUTH_TYPE", Elasticsearch_v2.BASIC_AUTH)
        mocker.patch("Elasticsearch_v2.USERNAME", None)
        mocker.patch("Elasticsearch_v2.PASSWORD", None)

        with pytest.raises(DemistoException):
            Elasticsearch_v2.get_kibana_auth_headers()


class TestBuildKibanaPath:
    """Tests for the build_kibana_path function."""

    def test_without_space_id(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.build_kibana_path("/api/cases") == "/api/cases"

    def test_without_leading_slash(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.build_kibana_path("api/cases") == "/api/cases"

    def test_with_space_id(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.build_kibana_path("/api/cases", space_id="my-space") == "/s/my-space/api/cases"


class TestKibanaHttpRequest:
    """Tests for the kibana_http_request function."""

    @pytest.fixture(autouse=True)
    def setup(self, mocker):
        mocker.patch("Elasticsearch_v2.SERVER", "https://my-deployment.es.us-central1.gcp.cloud.es.io")
        mocker.patch("Elasticsearch_v2.get_kibana_auth_headers", return_value={"Authorization": "Basic dXNlcjpwYXNz"})
        mocker.patch("Elasticsearch_v2.INSECURE", True)
        mocker.patch("Elasticsearch_v2.TIMEOUT", 60)
        mocker.patch("Elasticsearch_v2.DEFAULT_SPACE_ID", "")

    def test_get_request_success(self, mocker):
        """
        Given:
            - A successful GET response from Kibana
        When:
            - Calling kibana_http_request
        Then:
            - Return the parsed JSON response
        """
        import Elasticsearch_v2

        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.content = b'{"id": "123"}'
        mock_response.json.return_value = {"id": "123"}
        mock_request = mocker.patch("Elasticsearch_v2.requests.request", return_value=mock_response)

        result = Elasticsearch_v2.kibana_http_request("GET", "/api/cases/123")

        assert result == {"id": "123"}
        call_kwargs = mock_request.call_args[1]
        assert call_kwargs["url"] == "https://my-deployment.kb.us-central1.gcp.cloud.es.io/api/cases/123"
        assert "kbn-xsrf" not in call_kwargs["headers"]

    def test_post_request_adds_xsrf_header(self, mocker):
        """
        Given:
            - A POST request to Kibana
        When:
            - Calling kibana_http_request
        Then:
            - The kbn-xsrf header is added to the request
        """
        import Elasticsearch_v2

        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.content = b"{}"
        mock_response.json.return_value = {}
        mock_request = mocker.patch("Elasticsearch_v2.requests.request", return_value=mock_response)

        Elasticsearch_v2.kibana_http_request("POST", "/api/cases", json_data={"title": "test"})

        call_kwargs = mock_request.call_args[1]
        assert call_kwargs["headers"]["kbn-xsrf"] == "true"
        assert call_kwargs["json"] == {"title": "test"}

    def test_space_id_prefixes_path(self, mocker):
        """
        Given:
            - A space_id argument
        When:
            - Calling kibana_http_request
        Then:
            - The request URL includes the "/s/{space_id}" prefix
        """
        import Elasticsearch_v2

        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.content = b"{}"
        mock_response.json.return_value = {}
        mock_request = mocker.patch("Elasticsearch_v2.requests.request", return_value=mock_response)

        Elasticsearch_v2.kibana_http_request("GET", "/api/cases", space_id="my-space")

        call_kwargs = mock_request.call_args[1]
        assert call_kwargs["url"] == "https://my-deployment.kb.us-central1.gcp.cloud.es.io/s/my-space/api/cases"

    def test_default_space_id_used_when_not_provided(self, mocker):
        """
        Given:
            - No explicit space_id argument, but a configured DEFAULT_SPACE_ID
        When:
            - Calling kibana_http_request
        Then:
            - The request URL includes the default space prefix
        """
        import Elasticsearch_v2

        mocker.patch("Elasticsearch_v2.DEFAULT_SPACE_ID", "default-space")
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.content = b"{}"
        mock_response.json.return_value = {}
        mock_request = mocker.patch("Elasticsearch_v2.requests.request", return_value=mock_response)

        Elasticsearch_v2.kibana_http_request("GET", "/api/cases")

        call_kwargs = mock_request.call_args[1]
        assert "/s/default-space/api/cases" in call_kwargs["url"]

    def test_error_status_code_raises(self, mocker):
        """
        Given:
            - A Kibana response with a non-ok status code and a JSON error body
        When:
            - Calling kibana_http_request
        Then:
            - Raise a DemistoException containing the error message
        """
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        mock_response = MagicMock()
        mock_response.status_code = 404
        mock_response.content = b'{"message": "Case not found"}'
        mock_response.json.return_value = {"message": "Case not found"}
        mock_response.text = '{"message": "Case not found"}'
        mocker.patch("Elasticsearch_v2.requests.request", return_value=mock_response)

        with pytest.raises(DemistoException) as exc_info:
            Elasticsearch_v2.kibana_http_request("GET", "/api/cases/unknown")

        assert "Case not found" in str(exc_info.value)

    def test_connection_error_raises_demisto_exception(self, mocker):
        """
        Given:
            - requests.request raises a connection error
        When:
            - Calling kibana_http_request
        Then:
            - Raise a DemistoException
        """
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        mocker.patch("Elasticsearch_v2.requests.request", side_effect=requests.exceptions.ConnectionError("boom"))

        with pytest.raises(DemistoException) as exc_info:
            Elasticsearch_v2.kibana_http_request("GET", "/api/cases")

        assert "Failed connecting to Kibana" in str(exc_info.value)

    def test_empty_response_returns_empty_dict(self, mocker):
        """
        Given:
            - A 204 No Content response
        When:
            - Calling kibana_http_request
        Then:
            - Return an empty dict
        """
        import Elasticsearch_v2

        mock_response = MagicMock()
        mock_response.status_code = 204
        mock_response.content = b""
        mocker.patch("Elasticsearch_v2.requests.request", return_value=mock_response)

        result = Elasticsearch_v2.kibana_http_request("DELETE", "/api/cases/123")

        assert result == {}


class TestGetJsonBodyFromEntryId:
    """Tests for the get_json_body_from_entry_id function."""

    def test_valid_json_file(self, mocker, tmp_path):
        """
        Given:
            - An entry_id pointing to a valid JSON file
        When:
            - Calling get_json_body_from_entry_id
        Then:
            - Return the parsed JSON content
        """
        import Elasticsearch_v2

        file_path = tmp_path / "body.json"
        file_path.write_text(json.dumps({"title": "test case"}))
        mocker.patch("Elasticsearch_v2.demisto.getFilePath", return_value={"path": str(file_path), "name": "body.json"})

        result = Elasticsearch_v2.get_json_body_from_entry_id("123@abc")

        assert result == {"title": "test case"}

    def test_invalid_json_raises(self, mocker, tmp_path):
        """
        Given:
            - An entry_id pointing to a file with invalid JSON content
        When:
            - Calling get_json_body_from_entry_id
        Then:
            - Raise a DemistoException
        """
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        file_path = tmp_path / "body.json"
        file_path.write_text("not valid json")
        mocker.patch("Elasticsearch_v2.demisto.getFilePath", return_value={"path": str(file_path), "name": "body.json"})

        with pytest.raises(DemistoException) as exc_info:
            Elasticsearch_v2.get_json_body_from_entry_id("123@abc")

        assert "does not contain valid JSON" in str(exc_info.value)

    def test_missing_file_path_raises(self, mocker):
        """
        Given:
            - demisto.getFilePath returns no path
        When:
            - Calling get_json_body_from_entry_id
        Then:
            - Raise a DemistoException
        """
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        mocker.patch("Elasticsearch_v2.demisto.getFilePath", return_value={})

        with pytest.raises(DemistoException) as exc_info:
            Elasticsearch_v2.get_json_body_from_entry_id("bad-entry-id")

        assert "Could not resolve file path" in str(exc_info.value)

    def test_get_file_path_exception_raises(self, mocker):
        """
        Given:
            - demisto.getFilePath raises an exception (e.g. entry not found)
        When:
            - Calling get_json_body_from_entry_id
        Then:
            - Raise a DemistoException
        """
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        mocker.patch("Elasticsearch_v2.demisto.getFilePath", side_effect=Exception("not found"))

        with pytest.raises(DemistoException) as exc_info:
            Elasticsearch_v2.get_json_body_from_entry_id("bad-entry-id")

        assert "Failed to retrieve file info" in str(exc_info.value)


MOCK_KIBANA_CASE = {
    "id": "case-id-1",
    "title": "Test Case",
    "description": "A test case",
    "owner": "cases",
    "severity": "medium",
    "status": "open",
    "created_at": "2024-01-01T00:00:00.000Z",
    "connector": {"type": ".none"},
}


class TestBuildCaseConnector:
    """Tests for build_case_connector and build_case_connector_fields."""

    def test_raw_connector_fields_json_string_takes_precedence(self):
        import Elasticsearch_v2

        args = {
            "connector_id": "conn1",
            "connector_fields": '{"custom": "value"}',
            "connector_fields_priority_jira": "High",
        }
        connector = Elasticsearch_v2.build_case_connector(args)

        assert connector["fields"] == {"custom": "value"}
        assert connector["id"] == "conn1"

    def test_flattened_jira_fields(self):
        import Elasticsearch_v2

        args = {
            "connector_type": ".jira",
            "connector_fields_issue_type_jira": "Bug",
            "connector_fields_priority_jira": "High",
        }
        connector = Elasticsearch_v2.build_case_connector(args)

        assert connector["type"] == ".jira"
        assert connector["fields"] == {"issueType": "Bug", "priority": "High"}

    def test_servicenow_boolean_fields(self):
        import Elasticsearch_v2

        args = {
            "connector_type": ".servicenow",
            "connector_fields_dest_ip_servicenow": "true",
            "connector_fields_malware_hash_servicenow": "false",
        }
        connector = Elasticsearch_v2.build_case_connector(args)

        assert connector["fields"] == {"destIp": True, "malwareHash": False}

    def test_no_connector_args_returns_none(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.build_case_connector({}) is None


class TestBuildCaseBody:
    """Tests for build_case_body."""

    def test_builds_basic_fields(self):
        import Elasticsearch_v2

        args = {"title": "My Case", "description": "desc", "owner": "cases", "severity": "high", "tags": "a,b"}
        body = Elasticsearch_v2.build_case_body(args)

        assert body["title"] == "My Case"
        assert body["owner"] == "cases"
        assert body["tags"] == ["a", "b"]

    def test_require_owner_raises_when_missing(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.build_case_body({"title": "x"}, require_owner=True)

    def test_assignees_built_from_uid_list(self):
        import Elasticsearch_v2

        body = Elasticsearch_v2.build_case_body({"assignee_uid": "uid1,uid2"})

        assert body["assignees"] == [{"uid": "uid1"}, {"uid": "uid2"}]

    def test_settings_built_from_sync_and_extract(self):
        import Elasticsearch_v2

        body = Elasticsearch_v2.build_case_body({"sync_alerts": "true", "extract_observables": "false"})

        assert body["settings"] == {"syncAlerts": True, "extractObservables": False}


class TestEsKibanaCaseCreateCommand:
    """Tests for es_kibana_case_create_command."""

    def test_create_case_success(self, mocker):
        """
        Given:
            - Arguments to create a Kibana case
        When:
            - Calling es_kibana_case_create_command
        Then:
            - POST /api/cases is called and a CommandResults with the case data is returned
        """
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_KIBANA_CASE)

        result = Elasticsearch_v2.es_kibana_case_create_command(
            {"title": "Test Case", "owner": "cases", "description": "A test case"}, {}
        )

        assert result.outputs == MOCK_KIBANA_CASE
        assert result.outputs_prefix == "Elasticsearch.Kibana.Case"
        assert result.outputs_key_field == "id"
        call_args = mock_request.call_args
        assert call_args[0][0] == "POST"
        assert call_args[0][1] == "/api/cases"

    def test_create_case_with_entry_id_overrides_args(self, mocker):
        """
        Given:
            - An entry_id argument along with other case arguments
        When:
            - Calling es_kibana_case_create_command
        Then:
            - The JSON body from the file is used instead of the individual arguments
        """
        import Elasticsearch_v2

        mocker.patch("Elasticsearch_v2.get_json_body_from_entry_id", return_value={"title": "From File"})
        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_KIBANA_CASE)

        Elasticsearch_v2.es_kibana_case_create_command({"entry_id": "123@abc", "title": "Ignored"}, {})

        call_args = mock_request.call_args
        assert call_args[1]["json_data"] == {"title": "From File"}

    def test_create_case_missing_owner_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_case_create_command({"title": "Test Case"}, {})


class TestEsKibanaCaseUpdateCommand:
    """Tests for es_kibana_case_update_command."""

    def test_update_case_success(self, mocker):
        """
        Given:
            - case_id, version and fields to update
        When:
            - Calling es_kibana_case_update_command
        Then:
            - PATCH /api/cases is called with a "cases" array payload
        """
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=[MOCK_KIBANA_CASE])

        result = Elasticsearch_v2.es_kibana_case_update_command({"case_id": "case-id-1", "version": "v1", "status": "closed"}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "PATCH"
        payload = call_args[1]["json_data"]
        assert payload["cases"][0]["id"] == "case-id-1"
        assert payload["cases"][0]["version"] == "v1"
        assert payload["cases"][0]["status"] == "closed"
        assert result.outputs == MOCK_KIBANA_CASE

    def test_update_case_missing_case_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_case_update_command({"version": "v1"}, {})

    def test_update_case_missing_version_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_case_update_command({"case_id": "case-id-1"}, {})


class TestEsKibanaCaseDeleteCommand:
    """Tests for es_kibana_case_delete_command."""

    def test_delete_case_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        result = Elasticsearch_v2.es_kibana_case_delete_command({"case_id": "case-id-1,case-id-2"}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "DELETE"
        assert "case-id-1" in result.readable_output
        assert "case-id-2" in result.readable_output

    def test_delete_case_missing_case_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_case_delete_command({}, {})


class TestEsKibanaCaseListCommand:
    """Tests for es_kibana_case_list_command."""

    def test_list_by_case_id(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_KIBANA_CASE)

        result = Elasticsearch_v2.es_kibana_case_list_command({"case_id": "case-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/cases/case-id-1"
        assert result.outputs == [MOCK_KIBANA_CASE]

    def test_list_search(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={"cases": [MOCK_KIBANA_CASE]})

        result = Elasticsearch_v2.es_kibana_case_list_command({"status": "open", "page": "1", "size": "20"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/cases/_find"
        assert call_args[1]["params"]["status"] == "open"
        assert call_args[1]["params"]["perPage"] == "20"
        assert result.outputs == [MOCK_KIBANA_CASE]


class TestEsKibanaCaseAlertsListCommand:
    """Tests for es_kibana_case_alerts_list_command."""

    def test_list_alerts_success(self, mocker):
        import Elasticsearch_v2

        alerts = [{"id": "alert-1", "index": "idx1", "attached_at": "2024-01-01T00:00:00.000Z"}]
        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=alerts)

        result = Elasticsearch_v2.es_kibana_case_alerts_list_command({"case_id": "case-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/cases/case-id-1/alerts"
        assert result.outputs == alerts
        assert result.outputs_prefix == "Elasticsearch.Kibana.Case.case-id-1.Alert"

    def test_missing_case_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_case_alerts_list_command({}, {})


class TestEsKibanaCaseCommentCommands:
    """Tests for es_kibana_case_comment_add_command, update_command, and delete_command."""

    def test_comment_add_user_type(self, mocker):
        import Elasticsearch_v2

        response = {
            "id": "case-id-1",
            "comments": [{"comment": "hello", "created_by": {"username": "bob"}}],
        }
        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=response)

        result = Elasticsearch_v2.es_kibana_case_comment_add_command(
            {"case_id": "case-id-1", "owner": "cases", "type": "user", "comment": "hello"}, {}
        )

        call_args = mock_request.call_args
        assert call_args[0][0] == "POST"
        assert call_args[0][1] == "/api/cases/case-id-1/comments"
        assert call_args[1]["json_data"]["comment"] == "hello"
        assert result.outputs == response

    def test_comment_add_alert_type(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={"id": "case-id-1", "comments": []})

        Elasticsearch_v2.es_kibana_case_comment_add_command(
            {"case_id": "case-id-1", "owner": "securitySolution", "type": "alert", "alert_id": "a1", "index": "i1"}, {}
        )

        call_args = mock_request.call_args
        body = call_args[1]["json_data"]
        assert body["alertId"] == "a1"
        assert body["index"] == "i1"

    def test_comment_add_missing_type_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_case_comment_add_command({"case_id": "case-id-1", "owner": "cases"}, {})

    def test_comment_update_success(self, mocker):
        import Elasticsearch_v2

        response = {
            "id": "case-id-1",
            "comments": [{"id": "comment-1", "comment": "updated", "updated_by": {"username": "bob"}}],
        }
        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=response)

        result = Elasticsearch_v2.es_kibana_case_comment_update_command(
            {
                "case_id": "case-id-1",
                "owner": "cases",
                "type": "user",
                "comment": "updated",
                "comment_id": "comment-1",
                "version": "v1",
            },
            {},
        )

        call_args = mock_request.call_args
        assert call_args[0][0] == "PATCH"
        assert call_args[1]["json_data"]["id"] == "comment-1"
        assert result.outputs == response

    def test_comment_delete_success(self, mocker):
        import Elasticsearch_v2

        mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        result = Elasticsearch_v2.es_kibana_case_comment_delete_command({"case_id": "case-id-1"}, {})

        assert "case-id-1" in result.readable_output

    def test_comment_delete_missing_case_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_case_comment_delete_command({}, {})


class TestEsKibanaCaseFileAttachCommand:
    """Tests for es_kibana_case_file_attach_command."""

    def test_attach_file_success(self, mocker, tmp_path):
        import Elasticsearch_v2

        file_path = tmp_path / "report.pdf"
        file_path.write_bytes(b"%PDF-1.4 fake content")
        mocker.patch("Elasticsearch_v2.demisto.getFilePath", return_value={"path": str(file_path), "name": "report.pdf"})
        response = {"id": "case-id-1", "comments": [{"updated_by": {"username": "bob"}}]}
        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=response)

        result = Elasticsearch_v2.es_kibana_case_file_attach_command({"case_id": "case-id-1", "entry_id": "123@abc"}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "POST"
        assert call_args[0][1] == "/api/cases/case-id-1/files"
        assert "files" in call_args[1]
        assert result.outputs == response

    def test_attach_file_missing_case_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_case_file_attach_command({"entry_id": "123@abc"}, {})

    def test_attach_file_missing_entry_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_case_file_attach_command({"case_id": "case-id-1"}, {})


MOCK_ALERTING_HEALTH_RESPONSE = {
    "is_sufficiently_secure": True,
    "has_permanent_encryption_key": True,
    "alerting_framework_health": {
        "decryption_health": {"status": "ok"},
        "execution_health": {"status": "ok"},
        "read_health": {"status": "ok"},
    },
}

MOCK_RULE = {
    "id": "rule-id-1",
    "enabled": True,
    "name": "Test Rule",
    "rule_type_id": ".index-threshold",
    "created_at": "2024-01-01T00:00:00.000Z",
}


class TestEsKibanaAlertingHealthGetCommand:
    """Tests for es_kibana_alerting_health_get_command."""

    def test_get_health_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_ALERTING_HEALTH_RESPONSE)

        result = Elasticsearch_v2.es_kibana_alerting_health_get_command({}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "GET"
        assert call_args[0][1] == "/api/alerting/_health"
        assert result.outputs == MOCK_ALERTING_HEALTH_RESPONSE
        assert result.outputs_prefix == "Elasticsearch.Kibana.AlertingHealth"
        assert "Is sufficiently secure" in result.readable_output


class TestEsKibanaRuleTypesListCommand:
    """Tests for es_kibana_rule_types_list_command."""

    def test_list_rule_types_success(self, mocker):
        import Elasticsearch_v2

        rule_types = [{"id": "type1", "name": "Type 1", "category": "cat", "producer": "prod", "action_groups": [{"id": "ag1"}]}]
        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=rule_types)

        result = Elasticsearch_v2.es_kibana_rule_types_list_command({}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/alerting/rule_types"
        assert result.outputs == rule_types
        assert result.outputs_prefix == "Elasticsearch.Kibana.RuleTypes"


class TestEsKibanaRuleListCommand:
    """Tests for es_kibana_rule_list_command."""

    def test_list_by_rule_id(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_RULE)

        result = Elasticsearch_v2.es_kibana_rule_list_command({"rule_id": "rule-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/alerting/rule/rule-id-1"
        assert result.outputs == [MOCK_RULE]

    def test_list_search(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={"data": [MOCK_RULE]})

        result = Elasticsearch_v2.es_kibana_rule_list_command({"search": "test", "page": "1", "size": "10"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/alerting/rules/_find"
        assert call_args[1]["params"]["search"] == "test"
        assert result.outputs == [MOCK_RULE]


class TestEsKibanaRuleEnableDisableCommands:
    """Tests for es_kibana_rule_enable_command and es_kibana_rule_disable_command."""

    def test_enable_rule_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        result = Elasticsearch_v2.es_kibana_rule_enable_command({"rule_id": "rule-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "POST"
        assert call_args[0][1] == "/api/alerting/rule/rule-id-1/_enable"
        assert "rule-id-1" in result.readable_output
        assert "enabled" in result.readable_output

    def test_enable_rule_missing_rule_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_rule_enable_command({}, {})

    def test_disable_rule_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        result = Elasticsearch_v2.es_kibana_rule_disable_command({"rule_id": "rule-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/alerting/rule/rule-id-1/_disable"
        assert "disabled" in result.readable_output

    def test_disable_rule_missing_rule_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_rule_disable_command({}, {})


class TestBuildRuleUpdateBody:
    """Tests for build_rule_update_body."""

    def test_basic_fields(self):
        import Elasticsearch_v2

        body = Elasticsearch_v2.build_rule_update_body({"name": "New name", "schedule_interval": "5m"})

        assert body["name"] == "New name"
        assert body["schedule"] == {"interval": "5m"}

    def test_flapping_fields_merged(self):
        import Elasticsearch_v2

        body = Elasticsearch_v2.build_rule_update_body(
            {"flapping_enabled": "true", "flapping_look_back_window": "5", "flapping_status_change_threshold": "3"}
        )

        assert body["flapping"] == {"enabled": True, "look_back_window": 5, "status_change_threshold": 3}

    def test_artifacts_fields(self):
        import Elasticsearch_v2

        body = Elasticsearch_v2.build_rule_update_body(
            {"artifacts_dashboards_id": "d1,d2", "artifacts_investigation_guide_blob": "guide text"}
        )

        assert body["artifacts"]["dashboards"] == [{"id": "d1"}, {"id": "d2"}]
        assert body["artifacts"]["investigation_guide"] == {"blob": "guide text"}


class TestEsKibanaRuleUpdateCommand:
    """Tests for es_kibana_rule_update_command."""

    def test_update_rule_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_RULE)

        result = Elasticsearch_v2.es_kibana_rule_update_command({"rule_id": "rule-id-1", "name": "Updated"}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "PUT"
        assert call_args[0][1] == "/api/alerting/rule/rule-id-1"
        assert call_args[1]["json_data"] == {"name": "Updated"}
        assert result.outputs == MOCK_RULE
        assert "rule-id-1" in result.readable_output

    def test_update_rule_with_entry_id(self, mocker):
        import Elasticsearch_v2

        mocker.patch("Elasticsearch_v2.get_json_body_from_entry_id", return_value={"name": "From File"})
        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_RULE)

        Elasticsearch_v2.es_kibana_rule_update_command({"rule_id": "rule-id-1", "entry_id": "123@abc"}, {})

        call_args = mock_request.call_args
        assert call_args[1]["json_data"] == {"name": "From File"}

    def test_update_rule_missing_rule_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_rule_update_command({}, {})


class TestEsKibanaRuleAlertMuteUnmuteCommands:
    """Tests for es_kibana_rule_alert_mute_command and es_kibana_rule_alert_unmute_command."""

    def test_mute_specific_alert(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        result = Elasticsearch_v2.es_kibana_rule_alert_mute_command({"rule_id": "rule-id-1", "alert_id": "alert-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/alerting/rule/rule-id-1/alert/alert-1/_mute"
        assert "alert-1" in result.readable_output

    def test_mute_all(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        Elasticsearch_v2.es_kibana_rule_alert_mute_command({"rule_id": "rule-id-1", "mute_all": "true"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/alerting/rule/rule-id-1/_mute_all"

    def test_mute_missing_alert_id_and_mute_all_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_rule_alert_mute_command({"rule_id": "rule-id-1"}, {})

    def test_mute_missing_rule_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_rule_alert_mute_command({"alert_id": "alert-1"}, {})

    def test_unmute_specific_alert(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        result = Elasticsearch_v2.es_kibana_rule_alert_unmute_command({"rule_id": "rule-id-1", "alert_id": "alert-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/alerting/rule/rule-id-1/alert/alert-1/_unmute"
        assert "alert-1" in result.readable_output

    def test_unmute_all(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        Elasticsearch_v2.es_kibana_rule_alert_unmute_command({"rule_id": "rule-id-1", "unmute_all": "true"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/alerting/rule/rule-id-1/_unmute_all"


class TestEsKibanaDetectionAlertStatusSetCommand:
    """Tests for es_kibana_detection_alert_status_set_command."""

    def test_set_status_by_signal_ids(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={"total": 2, "updated": 2})
        mocker.patch("Elasticsearch_v2.safe_load_json", return_value=None)

        result = Elasticsearch_v2.es_kibana_detection_alert_status_set_command({"status": "closed", "signal_ids": "id1,id2"}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "POST"
        assert call_args[0][1] == "/api/detection_engine/signals/status"
        assert call_args[1]["json_data"]["signal_ids"] == ["id1", "id2"]
        assert result.outputs == {"total": 2, "updated": 2}

    def test_set_status_by_query(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={"total": 1, "updated": 1})

        Elasticsearch_v2.es_kibana_detection_alert_status_set_command({"status": "open", "query": '{"match_all": {}}'}, {})

        call_args = mock_request.call_args
        assert call_args[1]["json_data"]["query"] == {"match_all": {}}

    def test_missing_status_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_detection_alert_status_set_command({}, {})


MOCK_EXCEPTION_ITEM = {
    "id": "item-id-1",
    "item_id": "trusted-linux-processes",
    "list_id": "list-id-1",
    "name": "Test Item",
    "description": "A test item",
    "created_at": "2024-01-01T00:00:00.000Z",
}

MOCK_EXCEPTION_LIST = {
    "id": "list-id-1",
    "list_id": "trusted-linux-processes",
    "name": "Test List",
    "description": "A test list",
    "created_at": "2024-01-01T00:00:00.000Z",
}


class TestBuildExceptionEntry:
    """Tests for build_exception_entry."""

    def test_no_entries_returns_none(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.build_exception_entry({}) is None

    def test_simple_value_entry(self):
        import Elasticsearch_v2

        entry = Elasticsearch_v2.build_exception_entry(
            {"entries_field": "file.path", "entries_type": "match", "entries_operator": "included", "entries_value": "/bin/bash"}
        )

        assert entry == {"field": "file.path", "type": "match", "operator": "included", "value": "/bin/bash"}

    def test_list_type_entry(self):
        import Elasticsearch_v2

        entry = Elasticsearch_v2.build_exception_entry(
            {
                "entries_field": "file.hash",
                "entries_type": "list",
                "entries_list_id": "list1",
                "entries_list_type": "keyword",
            }
        )

        assert entry["list"] == {"id": "list1", "type": "keyword"}


class TestEsKibanaEndpointExceptionListItemCreateCommand:
    """Tests for es_kibana_endpoint_exception_list_item_create_command."""

    def test_create_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_EXCEPTION_ITEM)

        result = Elasticsearch_v2.es_kibana_endpoint_exception_list_item_create_command(
            {"name": "Test Item", "description": "A test item"}, {}
        )

        call_args = mock_request.call_args
        assert call_args[0][0] == "POST"
        assert call_args[0][1] == "/api/endpoint_list/items"
        assert call_args[1]["json_data"]["type"] == "simple"
        assert result.outputs == MOCK_EXCEPTION_ITEM
        assert result.outputs_prefix == "Elasticsearch.Kibana.EndpointExceptionListItem"

    def test_create_with_entry_id(self, mocker):
        import Elasticsearch_v2

        mocker.patch("Elasticsearch_v2.get_json_body_from_entry_id", return_value={"name": "From File"})
        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_EXCEPTION_ITEM)

        Elasticsearch_v2.es_kibana_endpoint_exception_list_item_create_command({"entry_id": "123@abc"}, {})

        call_args = mock_request.call_args
        assert call_args[1]["json_data"] == {"name": "From File"}


class TestEsKibanaEndpointExceptionListItemUpdateCommand:
    """Tests for es_kibana_endpoint_exception_list_item_update_command."""

    def test_update_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_EXCEPTION_ITEM)

        result = Elasticsearch_v2.es_kibana_endpoint_exception_list_item_update_command(
            {"exception_list_item_id": "item-id-1", "_version": "v1", "name": "Updated"}, {}
        )

        call_args = mock_request.call_args
        assert call_args[0][0] == "PUT"
        assert call_args[1]["json_data"]["id"] == "item-id-1"
        assert call_args[1]["json_data"]["_version"] == "v1"
        assert result.outputs == MOCK_EXCEPTION_ITEM


class TestEsKibanaEndpointExceptionListItemDeleteCommand:
    """Tests for es_kibana_endpoint_exception_list_item_delete_command."""

    def test_delete_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        result = Elasticsearch_v2.es_kibana_endpoint_exception_list_item_delete_command(
            {"item_id": "trusted-linux-processes"}, {}
        )

        call_args = mock_request.call_args
        assert call_args[0][0] == "DELETE"
        assert "trusted-linux-processes" in result.readable_output

    def test_missing_item_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_endpoint_exception_list_item_delete_command({}, {})


class TestEsKibanaEndpointExceptionListItemListCommand:
    """Tests for es_kibana_endpoint_exception_list_item_list_command."""

    def test_list_by_item_id(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_EXCEPTION_ITEM)

        result = Elasticsearch_v2.es_kibana_endpoint_exception_list_item_list_command({"item_id": "trusted-linux-processes"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/endpoint_list/items"
        assert result.outputs == [MOCK_EXCEPTION_ITEM]

    def test_list_find(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={"data": [MOCK_EXCEPTION_ITEM]})

        result = Elasticsearch_v2.es_kibana_endpoint_exception_list_item_list_command({"page": "1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/endpoint_list/items/_find"
        assert result.outputs == [MOCK_EXCEPTION_ITEM]


class TestEsKibanaExceptionListListCommand:
    """Tests for es_kibana_exception_list_list_command."""

    def test_list_by_id(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_EXCEPTION_LIST)

        result = Elasticsearch_v2.es_kibana_exception_list_list_command({"exception_list_id": "list-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/exception_lists"
        assert result.outputs == [MOCK_EXCEPTION_LIST]

    def test_list_find(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={"data": [MOCK_EXCEPTION_LIST]})

        result = Elasticsearch_v2.es_kibana_exception_list_list_command({}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/exception_lists/_find"
        assert result.outputs == [MOCK_EXCEPTION_LIST]


class TestEsKibanaExceptionListCreateCommand:
    """Tests for es_kibana_exception_list_create_command."""

    def test_create_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_EXCEPTION_LIST)

        result = Elasticsearch_v2.es_kibana_exception_list_create_command({"name": "Test List", "type": "detection"}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "POST"
        assert call_args[1]["json_data"]["type"] == "detection"
        assert result.outputs == MOCK_EXCEPTION_LIST

    def test_missing_type_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_exception_list_create_command({"name": "Test List"}, {})


class TestEsKibanaExceptionListUpdateCommand:
    """Tests for es_kibana_exception_list_update_command."""

    def test_update_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_EXCEPTION_LIST)

        result = Elasticsearch_v2.es_kibana_exception_list_update_command(
            {"description": "desc", "name": "Test List", "type": "detection", "exception_list_id": "list-id-1"}, {}
        )

        call_args = mock_request.call_args
        assert call_args[0][0] == "PUT"
        assert call_args[1]["json_data"]["id"] == "list-id-1"
        assert result.outputs == MOCK_EXCEPTION_LIST

    def test_missing_required_field_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_exception_list_update_command({"name": "Test List", "type": "detection"}, {})


class TestEsKibanaExceptionListDeleteCommand:
    """Tests for es_kibana_exception_list_delete_command."""

    def test_delete_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        result = Elasticsearch_v2.es_kibana_exception_list_delete_command({"exception_list_id": "list-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "DELETE"
        assert "list-id-1" in result.readable_output

    def test_missing_identifiers_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_exception_list_delete_command({}, {})


class TestEsKibanaExceptionListItemListCommand:
    """Tests for es_kibana_exception_list_item_list_command."""

    def test_list_by_id(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_EXCEPTION_ITEM)

        result = Elasticsearch_v2.es_kibana_exception_list_item_list_command({"exception_list_item_id": "item-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/exception_lists/items"
        assert result.outputs == [MOCK_EXCEPTION_ITEM]

    def test_list_find(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={"data": [MOCK_EXCEPTION_ITEM]})

        result = Elasticsearch_v2.es_kibana_exception_list_item_list_command({"search": "test"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/exception_lists/items/_find"
        assert result.outputs == [MOCK_EXCEPTION_ITEM]


class TestEsKibanaExceptionListItemCreateCommand:
    """Tests for es_kibana_exception_list_item_create_command."""

    def test_create_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_EXCEPTION_ITEM)

        result = Elasticsearch_v2.es_kibana_exception_list_item_create_command({"name": "Test Item", "list_id": "list-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "POST"
        assert call_args[0][1] == "/api/exception_lists/items"
        assert result.outputs == MOCK_EXCEPTION_ITEM


class TestEsKibanaExceptionItemListUpdateCommand:
    """Tests for es_kibana_exception_item_list_update_command."""

    def test_update_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_EXCEPTION_ITEM)

        result = Elasticsearch_v2.es_kibana_exception_item_list_update_command(
            {"exception_list_item_id": "item-id-1", "name": "Updated"}, {}
        )

        call_args = mock_request.call_args
        assert call_args[0][0] == "PUT"
        assert call_args[1]["json_data"]["id"] == "item-id-1"
        assert result.outputs == MOCK_EXCEPTION_ITEM


class TestEsKibanaExceptionListItemDeleteCommand:
    """Tests for es_kibana_exception_list_item_delete_command."""

    def test_delete_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        result = Elasticsearch_v2.es_kibana_exception_list_item_delete_command({"exception_list_item_id": "item-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "DELETE"
        assert "item-id-1" in result.readable_output

    def test_missing_identifiers_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_exception_list_item_delete_command({}, {})


MOCK_VALUE_LIST = {
    "id": "value-list-id-1",
    "name": "Test Value List",
    "description": "A test value list",
    "created_at": "2024-01-01T00:00:00.000Z",
}

MOCK_VALUE_LIST_ITEM = {
    "id": "value-list-item-id-1",
    "list_id": "value-list-id-1",
    "name": "Test Item",
    "description": "A test value list item",
    "created_at": "2024-01-01T00:00:00.000Z",
}


class TestEsKibanaValueListsListCommand:
    """Tests for es_kibana_value_lists_list_command."""

    def test_list_by_id(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_VALUE_LIST)

        result = Elasticsearch_v2.es_kibana_value_lists_list_command({"value_list_id": "value-list-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/lists"
        assert result.outputs == [MOCK_VALUE_LIST]

    def test_list_find(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={"data": [MOCK_VALUE_LIST]})

        result = Elasticsearch_v2.es_kibana_value_lists_list_command({}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/lists/_find"
        assert result.outputs == [MOCK_VALUE_LIST]


class TestEsKibanaValueListItemGetCommand:
    """Tests for es_kibana_value_list_item_get_command."""

    def test_get_by_item_id(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_VALUE_LIST_ITEM)

        result = Elasticsearch_v2.es_kibana_value_list_item_get_command({"value_list_item_id": "item-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/lists/items"
        assert result.outputs == [MOCK_VALUE_LIST_ITEM]

    def test_get_by_value_only(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_VALUE_LIST_ITEM)

        Elasticsearch_v2.es_kibana_value_list_item_get_command({"value": "1.2.3.4"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/lists/items"
        assert call_args[1]["params"]["value"] == "1.2.3.4"

    def test_get_find(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={"data": [MOCK_VALUE_LIST_ITEM]})

        result = Elasticsearch_v2.es_kibana_value_list_item_get_command({"value_list_id": "value-list-id-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][1] == "/api/lists/items/_find"
        assert result.outputs == [MOCK_VALUE_LIST_ITEM]


class TestEsKibanaValueListItemCreateCommand:
    """Tests for es_kibana_value_list_item_create_command."""

    def test_create_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_VALUE_LIST_ITEM)

        result = Elasticsearch_v2.es_kibana_value_list_item_create_command(
            {"value_list_id": "value-list-id-1", "value": "1.2.3.4"}, {}
        )

        call_args = mock_request.call_args
        assert call_args[0][0] == "POST"
        assert call_args[1]["json_data"] == {"list_id": "value-list-id-1", "value": "1.2.3.4"}
        assert result.outputs == MOCK_VALUE_LIST_ITEM

    def test_missing_value_list_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_value_list_item_create_command({"value": "1.2.3.4"}, {})

    def test_missing_value_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_value_list_item_create_command({"value_list_id": "value-list-id-1"}, {})


class TestEsKibanaValueListItemUpdateCommand:
    """Tests for es_kibana_value_list_item_update_command."""

    def test_update_success(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=MOCK_VALUE_LIST_ITEM)

        result = Elasticsearch_v2.es_kibana_value_list_item_update_command(
            {"value_list_item_id": "item-1", "value": "5.6.7.8"}, {}
        )

        call_args = mock_request.call_args
        assert call_args[0][0] == "PUT"
        assert call_args[1]["json_data"]["id"] == "item-1"
        assert result.outputs == MOCK_VALUE_LIST_ITEM

    def test_missing_value_list_item_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_value_list_item_update_command({"value": "1.2.3.4"}, {})


class TestEsKibanaValueListItemDeleteCommand:
    """Tests for es_kibana_value_list_item_delete_command."""

    def test_delete_by_item_id(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        result = Elasticsearch_v2.es_kibana_value_list_item_delete_command({"value_list_item_id": "item-1"}, {})

        call_args = mock_request.call_args
        assert call_args[0][0] == "DELETE"
        assert "item-1" in result.readable_output

    def test_delete_by_list_id_and_value(self, mocker):
        import Elasticsearch_v2

        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value={})

        Elasticsearch_v2.es_kibana_value_list_item_delete_command({"value_list_id": "value-list-id-1", "value": "1.2.3.4"}, {})

        call_args = mock_request.call_args
        assert call_args[1]["params"]["list_id"] == "value-list-id-1"
        assert call_args[1]["params"]["value"] == "1.2.3.4"

    def test_missing_identifiers_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_value_list_item_delete_command({}, {})


class TestEsKibanaValueListItemExportCommand:
    """Tests for es_kibana_value_list_item_export_command."""

    def test_export_success(self, mocker):
        import Elasticsearch_v2

        mocker.patch("Elasticsearch_v2.kibana_http_request", return_value="1.2.3.4\n5.6.7.8")
        mocker.patch("Elasticsearch_v2.fileResult", return_value={"Type": 3, "File": "value-list-items.txt"})

        results = Elasticsearch_v2.es_kibana_value_list_item_export_command({"value_list_id": "value-list-id-1"}, {})

        assert isinstance(results, list)
        assert len(results) == 2
        assert results[0].readable_output == "Successful response"


class TestEsKibanaValueListItemImportCommand:
    """Tests for es_kibana_value_list_item_import_command."""

    def test_import_success(self, mocker, tmp_path):
        import Elasticsearch_v2

        file_path = tmp_path / "values.txt"
        file_path.write_text("1.2.3.4\n5.6.7.8")
        mocker.patch("Elasticsearch_v2.demisto.getFilePath", return_value={"path": str(file_path), "name": "values.txt"})
        mock_request = mocker.patch("Elasticsearch_v2.kibana_http_request", return_value=[MOCK_VALUE_LIST_ITEM])

        result = Elasticsearch_v2.es_kibana_value_list_item_import_command(
            {"entry_id": "123@abc", "value_list_id": "value-list-id-1", "type": "ip"}, {}
        )

        call_args = mock_request.call_args
        assert call_args[0][0] == "POST"
        assert call_args[0][1] == "/api/lists/items/_import"
        assert "files" in call_args[1]
        assert result.outputs == [MOCK_VALUE_LIST_ITEM]

    def test_missing_entry_id_raises(self):
        import Elasticsearch_v2
        from CommonServerPython import DemistoException

        with pytest.raises(DemistoException):
            Elasticsearch_v2.es_kibana_value_list_item_import_command({}, {})


class TestBuildFetchExtraParams:
    """Tests for build_fetch_extra_params.
    A blank "Fields to Fetch" parameter must not add an empty "fields": [] entry to the
    fetch request body, otherwise Elasticsearch fails with:
    ParsingException 400 'Unknown key for a START_ARRAY in [fields]'.
    """

    def test_empty_fields_list_omits_fields_key(self):
        import Elasticsearch_v2

        extra_params = Elasticsearch_v2.build_fetch_extra_params([])

        assert "fields" not in extra_params
        assert extra_params == {"_source": True}

    def test_populated_fields_list_includes_fields_key(self):
        import Elasticsearch_v2

        extra_params = Elasticsearch_v2.build_fetch_extra_params(["field_a", "field_b"])

        assert extra_params == {"_source": True, "fields": ["field_a", "field_b"]}


class TestElasticEntityKind:
    """Tests for the ``elastic_entity_kind`` field added to mirrored/fetched incidents.

    The field carries the source entity kind so the incoming mapper can distinguish
    security alerts from cases. Security alerts derive it from ``event.kind`` (falling
    back to ``signal``); cases derive it from the case ``owner`` (falling back to
    ``securitySolution``).
    """

    def test_get_remote_data_security_alert_uses_event_kind(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_SECURITY_ALERT)
        mocker.patch.object(Elasticsearch_v2, "CLOSE_INCIDENT", False)
        mocker.patch.object(Elasticsearch_v2.demisto, "integrationInstance", return_value="instance-1")
        mocker.patch.object(
            Elasticsearch_v2,
            "search_security_alerts",
            return_value={
                "hits": {
                    "hits": [
                        {
                            "_index": "alerts",
                            "_id": "alert-1",
                            "_source": {
                                "kibana.alert.uuid": "alert-1",
                                "kibana.alert.severity": "high",
                                "kibana.alert.workflow_status": "open",
                                "event.kind": "signal",
                            },
                        }
                    ]
                }
            },
        )

        response = Elasticsearch_v2.get_remote_data_command({"id": "alert-1", "lastUpdate": ""}, {})

        assert response.mirrored_object[Elasticsearch_v2.ELASTIC_ENTITY_KIND_FIELD] == "signal"

    def test_get_remote_data_security_alert_defaults_kind(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_SECURITY_ALERT)
        mocker.patch.object(Elasticsearch_v2, "CLOSE_INCIDENT", False)
        mocker.patch.object(Elasticsearch_v2.demisto, "integrationInstance", return_value="instance-1")
        mocker.patch.object(
            Elasticsearch_v2,
            "search_security_alerts",
            return_value={
                "hits": {
                    "hits": [
                        {
                            "_index": "alerts",
                            "_id": "alert-2",
                            "_source": {
                                "kibana.alert.uuid": "alert-2",
                                "kibana.alert.severity": "low",
                                "kibana.alert.workflow_status": "open",
                            },
                        }
                    ]
                }
            },
        )

        response = Elasticsearch_v2.get_remote_data_command({"id": "alert-2", "lastUpdate": ""}, {})

        assert response.mirrored_object[Elasticsearch_v2.ELASTIC_ENTITY_KIND_FIELD] == Elasticsearch_v2.ENTITY_KIND_SECURITY_ALERT

    def test_get_remote_data_security_alert_exposes_flat_workflow_tags(self, mocker):
        """Mirroring-in must expose workflow tags under a dot-free key the incoming mapper can read.

        Alerts-as-data documents store the tags under the flat dotted key
        ``kibana.alert.workflow_tags``; the mapper's ``simple`` selector would otherwise treat that
        as nested-object traversal and resolve nothing, so tags would never mirror in.
        """
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_SECURITY_ALERT)
        mocker.patch.object(Elasticsearch_v2, "CLOSE_INCIDENT", False)
        mocker.patch.object(Elasticsearch_v2.demisto, "integrationInstance", return_value="instance-1")
        mocker.patch.object(
            Elasticsearch_v2,
            "search_security_alerts",
            return_value={
                "hits": {
                    "hits": [
                        {
                            "_index": "alerts",
                            "_id": "alert-tags",
                            "_source": {
                                "kibana.alert.uuid": "alert-tags",
                                "kibana.alert.severity": "low",
                                "kibana.alert.workflow_status": "open",
                                "kibana.alert.workflow_tags": ["new tag", "Duplicate"],
                            },
                        }
                    ]
                }
            },
        )

        response = Elasticsearch_v2.get_remote_data_command({"id": "alert-tags", "lastUpdate": ""}, {})

        assert response.mirrored_object["kibana_alert_workflow_tags"] == ["new tag", "Duplicate"]

    def test_get_remote_data_security_alert_tags_default_empty(self, mocker):
        """When an alert has no workflow tags, mirroring-in exposes an empty list, not a missing key."""
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_SECURITY_ALERT)
        mocker.patch.object(Elasticsearch_v2, "CLOSE_INCIDENT", False)
        mocker.patch.object(Elasticsearch_v2.demisto, "integrationInstance", return_value="instance-1")
        mocker.patch.object(
            Elasticsearch_v2,
            "search_security_alerts",
            return_value={
                "hits": {
                    "hits": [
                        {
                            "_index": "alerts",
                            "_id": "alert-no-tags",
                            "_source": {
                                "kibana.alert.uuid": "alert-no-tags",
                                "kibana.alert.severity": "low",
                                "kibana.alert.workflow_status": "open",
                            },
                        }
                    ]
                }
            },
        )

        response = Elasticsearch_v2.get_remote_data_command({"id": "alert-no-tags", "lastUpdate": ""}, {})

        assert response.mirrored_object["kibana_alert_workflow_tags"] == []

    def test_get_remote_data_case_uses_owner(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_CASE)
        mocker.patch.object(Elasticsearch_v2, "CLOSE_INCIDENT", False)
        mocker.patch.object(Elasticsearch_v2, "FETCH_ALERTS_FOR_CASE", False)
        mocker.patch.object(Elasticsearch_v2.demisto, "integrationInstance", return_value="instance-1")
        mocker.patch.object(
            Elasticsearch_v2,
            "kibana_http_request",
            return_value={"id": "case-1", "status": "open", "severity": "medium", "owner": "observability"},
        )

        response = Elasticsearch_v2.get_remote_data_command({"id": "case-1", "lastUpdate": ""}, {})

        assert response.mirrored_object[Elasticsearch_v2.ELASTIC_ENTITY_KIND_FIELD] == "observability"

    def test_get_remote_data_case_defaults_kind(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_CASE)
        mocker.patch.object(Elasticsearch_v2, "CLOSE_INCIDENT", False)
        mocker.patch.object(Elasticsearch_v2, "FETCH_ALERTS_FOR_CASE", False)
        mocker.patch.object(Elasticsearch_v2.demisto, "integrationInstance", return_value="instance-1")
        mocker.patch.object(
            Elasticsearch_v2,
            "kibana_http_request",
            return_value={"id": "case-2", "status": "open", "severity": "medium"},
        )

        response = Elasticsearch_v2.get_remote_data_command({"id": "case-2", "lastUpdate": ""}, {})

        assert response.mirrored_object[Elasticsearch_v2.ELASTIC_ENTITY_KIND_FIELD] == Elasticsearch_v2.ENTITY_KIND_CASE


class TestMirroringHelpers:
    """Unit tests for the small, pure helpers introduced with the mirroring feature."""

    def test_get_value_by_dot_notation_nested(self):
        import Elasticsearch_v2

        source = {"kibana": {"alert": {"uuid": "abc"}}}
        assert Elasticsearch_v2.get_value_by_dot_notation(source, "kibana.alert.uuid") == "abc"

    def test_get_value_by_dot_notation_missing(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.get_value_by_dot_notation({"a": 1}, "a.b.c") is None

    def test_get_alert_source_value_prefers_flat_key(self):
        import Elasticsearch_v2

        source = {"kibana.alert.uuid": "flat", "kibana": {"alert": {"uuid": "nested"}}}
        assert Elasticsearch_v2.get_alert_source_value(source, "kibana.alert.uuid") == "flat"

    def test_get_alert_source_value_falls_back_to_nested(self):
        import Elasticsearch_v2

        source = {"kibana": {"alert": {"uuid": "nested"}}}
        assert Elasticsearch_v2.get_alert_source_value(source, "kibana.alert.uuid") == "nested"

    def test_get_alert_source_value_non_dict(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.get_alert_source_value("not-a-dict", "any.key") is None

    @pytest.mark.parametrize(
        "name,expected",
        [("critical", 4), ("high", 3), ("medium", 2), ("low", 1), ("CRITICAL", 4), ("unknown", 0), (None, 0)],
    )
    def test_convert_severity(self, name, expected):
        import Elasticsearch_v2

        assert Elasticsearch_v2.convert_severity(name) == expected

    @pytest.mark.parametrize(
        "value,expected",
        [(4, "critical"), (3, "high"), (2, "medium"), (1, "low"), ("high", "high"), ("HIGH", "high"), (0, None), ("bogus", None)],
    )
    def test_convert_severity_to_elastic(self, value, expected):
        import Elasticsearch_v2

        assert Elasticsearch_v2.convert_severity_to_elastic(value) == expected

    def test_flatten_alert_hit_merges_source_and_keeps_metadata(self):
        import Elasticsearch_v2

        hit = {"_index": "alerts", "_id": "1", "_source": {"a": 1, "b": 2}}
        flattened = Elasticsearch_v2.flatten_alert_hit(hit)
        assert flattened == {"_index": "alerts", "_id": "1", "a": 1, "b": 2}
        assert "_source" not in flattened

    def test_flatten_alert_hit_handles_missing_source(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.flatten_alert_hit({"_id": "1"}) == {"_id": "1"}

    def test_build_alert_lookup_body(self):
        import Elasticsearch_v2

        body = Elasticsearch_v2.build_alert_lookup_body("remote-1", size=5)
        should = body["query"]["bool"]["should"]
        assert {"term": {"kibana.alert.uuid": "remote-1"}} in should
        assert {"ids": {"values": ["remote-1"]}} in should
        assert body["query"]["bool"]["minimum_should_match"] == 1
        assert body["size"] == 5

    @pytest.mark.parametrize(
        "message,expected",
        [
            ("Error 429 Too Many Requests", True),
            ("hit the rate limit", True),
            ("too many requests", True),
            ("connection refused", False),
        ],
    )
    def test_is_rate_limit_error(self, message, expected):
        import Elasticsearch_v2

        assert Elasticsearch_v2._is_rate_limit_error(Exception(message)) is expected

    def test_parse_to_utc_valid(self):
        import Elasticsearch_v2

        parsed = Elasticsearch_v2.parse_to_utc("2020-01-01T00:00:00Z")
        assert parsed is not None
        assert parsed.tzinfo is not None

    def test_parse_to_utc_empty(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.parse_to_utc("") is None
        assert Elasticsearch_v2.parse_to_utc(None) is None


class TestIsIncidentClosing:
    """Tests for is_incident_closing."""

    def test_status_done_is_closing(self):
        import Elasticsearch_v2
        from CommonServerPython import IncidentStatus

        assert Elasticsearch_v2.is_incident_closing(IncidentStatus.DONE, {}) is True

    def test_close_delta_key_is_closing(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.is_incident_closing(None, {"closeReason": "Resolved"}) is True

    def test_not_closing(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.is_incident_closing(None, {"status": "open"}) is False

    def test_none_delta_not_closing(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.is_incident_closing(None, None) is False


class TestGetIncidentType:
    """Tests for get_incident_type."""

    def test_returns_type(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2.demisto, "incident", return_value={"type": "Elasticsearch Case"})
        assert Elasticsearch_v2.get_incident_type() == "Elasticsearch Case"

    def test_returns_empty_when_no_type(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2.demisto, "incident", return_value={})
        assert Elasticsearch_v2.get_incident_type() == ""

    def test_returns_empty_on_exception(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2.demisto, "incident", side_effect=Exception("no context"))
        assert Elasticsearch_v2.get_incident_type() == ""

    def test_returns_empty_when_not_dict(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2.demisto, "incident", return_value="not-a-dict")
        assert Elasticsearch_v2.get_incident_type() == ""


class TestGetConfiguredFetchIncidentType:
    """Tests for get_configured_fetch_incident_type."""

    def test_fetch_incident_type_param(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "PARAMS", {"fetch_incident_type": "Elasticsearch Case"})
        assert Elasticsearch_v2.get_configured_fetch_incident_type() == "Elasticsearch Case"

    def test_incident_type_param_fallback(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "PARAMS", {"incidentType": "Elasticsearch Security Alert"})
        assert Elasticsearch_v2.get_configured_fetch_incident_type() == "Elasticsearch Security Alert"

    def test_unrecognized_returns_empty(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "PARAMS", {"fetch_incident_type": "Something Else"})
        assert Elasticsearch_v2.get_configured_fetch_incident_type() == ""


class TestPruneFetchedIds:
    """Tests for prune_fetched_ids."""

    def test_keeps_recent_and_drops_expired(self):
        import Elasticsearch_v2
        from datetime import datetime, timedelta

        now = datetime(2020, 1, 30, tzinfo=UTC)
        fetched = {
            "recent": (now - timedelta(days=1)).isoformat(),
            "expired": (now - timedelta(days=Elasticsearch_v2.FETCHED_IDS_RETENTION_DAYS + 5)).isoformat(),
        }
        pruned = Elasticsearch_v2.prune_fetched_ids(fetched, now=now)
        assert "recent" in pruned
        assert "expired" not in pruned

    def test_keeps_unparsable(self):
        import Elasticsearch_v2
        from datetime import datetime

        now = datetime(2020, 1, 30, tzinfo=UTC)
        pruned = Elasticsearch_v2.prune_fetched_ids({"weird": "not-a-date"}, now=now)
        assert "weird" in pruned

    def test_empty_input(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.prune_fetched_ids({}) == {}
        assert Elasticsearch_v2.prune_fetched_ids(None) == {}


class TestResolveRemoteIncidentType:
    """Tests for resolve_remote_incident_type."""

    def test_empty_remote_id(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2.resolve_remote_incident_type("", {}) == ""

    def test_resolves_case(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "kibana_http_request", return_value={"id": "case-1"})
        assert Elasticsearch_v2.resolve_remote_incident_type("case-1", {}) == Elasticsearch_v2.INCIDENT_TYPE_CASE

    def test_resolves_alert(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "kibana_http_request", return_value=None)
        mocker.patch.object(Elasticsearch_v2, "search_security_alerts", return_value={"hits": {"hits": [{"_id": "a1"}]}})
        assert Elasticsearch_v2.resolve_remote_incident_type("a1", {}) == Elasticsearch_v2.INCIDENT_TYPE_SECURITY_ALERT

    def test_falls_back_to_configured_type(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "kibana_http_request", side_effect=Exception("down"))
        mocker.patch.object(Elasticsearch_v2, "search_security_alerts", side_effect=Exception("down"))
        mocker.patch.object(
            Elasticsearch_v2, "get_configured_fetch_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_CASE
        )
        assert Elasticsearch_v2.resolve_remote_incident_type("x", {}) == Elasticsearch_v2.INCIDENT_TYPE_CASE

    def test_matches_nothing(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "kibana_http_request", return_value=None)
        mocker.patch.object(Elasticsearch_v2, "search_security_alerts", return_value={"hits": {"hits": []}})
        mocker.patch.object(Elasticsearch_v2, "get_configured_fetch_incident_type", return_value="")
        assert Elasticsearch_v2.resolve_remote_incident_type("x", {}) == ""


class TestSearchSecurityAlerts:
    """Tests for search_security_alerts fallback behavior."""

    def test_falls_back_to_kibana(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "elasticsearch_builder", side_effect=Exception("es down"))
        kibana = mocker.patch.object(Elasticsearch_v2, "kibana_http_request", return_value={"hits": {"hits": [{"_id": "1"}]}})
        result = Elasticsearch_v2.search_security_alerts({"query": {}}, {})
        assert result == {"hits": {"hits": [{"_id": "1"}]}}
        assert kibana.called

    def test_raises_when_both_fail(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "elasticsearch_builder", side_effect=Exception("es down"))
        mocker.patch.object(Elasticsearch_v2, "kibana_http_request", side_effect=Exception("kibana down"))
        mocker.patch.object(Elasticsearch_v2, "mirror_error")
        with pytest.raises(Exception):
            Elasticsearch_v2.search_security_alerts({"query": {}}, {})


class TestGetRemoteDataClose:
    """Tests for the close-incident entries returned by get_remote_data_command."""

    def test_alert_closed_adds_close_entry(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_SECURITY_ALERT)
        mocker.patch.object(Elasticsearch_v2, "CLOSE_INCIDENT", True)
        mocker.patch.object(Elasticsearch_v2.demisto, "integrationInstance", return_value="inst")
        mocker.patch.object(
            Elasticsearch_v2,
            "search_security_alerts",
            return_value={
                "hits": {
                    "hits": [
                        {
                            "_id": "a1",
                            "_source": {
                                "kibana.alert.uuid": "a1",
                                "kibana.alert.workflow_status": "closed",
                                "kibana.alert.workflow_reason": "false_positive",
                                "kibana.alert.severity": "low",
                            },
                        }
                    ]
                }
            },
        )
        response = Elasticsearch_v2.get_remote_data_command({"id": "a1", "lastUpdate": ""}, {})
        assert any(e["Contents"].get("dbotIncidentClose") for e in response.entries)
        close_entry = response.entries[0]["Contents"]
        assert close_entry["closeReason"] == "false_positive"

    def test_case_closed_adds_close_entry(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_CASE)
        mocker.patch.object(Elasticsearch_v2, "CLOSE_INCIDENT", True)
        mocker.patch.object(Elasticsearch_v2, "FETCH_ALERTS_FOR_CASE", False)
        mocker.patch.object(Elasticsearch_v2.demisto, "integrationInstance", return_value="inst")
        mocker.patch.object(
            Elasticsearch_v2,
            "kibana_http_request",
            return_value={"id": "c1", "status": "closed", "severity": "low", "closeReason": "duplicate"},
        )
        response = Elasticsearch_v2.get_remote_data_command({"id": "c1", "lastUpdate": ""}, {})
        assert response.entries[0]["Contents"]["dbotIncidentClose"] is True
        assert response.entries[0]["Contents"]["closeReason"] == "duplicate"

    def test_case_rate_limit_raises(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_CASE)
        mocker.patch.object(Elasticsearch_v2, "kibana_http_request", side_effect=Exception("429 too many requests"))
        return_error = mocker.patch.object(Elasticsearch_v2, "return_error", side_effect=SystemExit)
        with pytest.raises(SystemExit):
            Elasticsearch_v2.get_remote_data_command({"id": "c1", "lastUpdate": ""}, {})
        assert return_error.called


class TestUpdateRemoteSystemCommand:
    """Tests for update_remote_system_command routing."""

    def test_no_remote_id_returns_empty(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_CASE)
        args = {"remoteId": "", "data": {}, "entries": [], "incidentChanged": True, "delta": {}, "status": 1}
        assert Elasticsearch_v2.update_remote_system_command(args, {}) == ""

    def test_routes_to_alert(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_SECURITY_ALERT)
        mirror_out = mocker.patch.object(Elasticsearch_v2, "_mirror_out_security_alert")
        args = {
            "remoteId": "a1",
            "data": {"type": Elasticsearch_v2.INCIDENT_TYPE_SECURITY_ALERT},
            "entries": [],
            "incidentChanged": True,
            "delta": {"status": "closed"},
            "status": 1,
        }
        assert Elasticsearch_v2.update_remote_system_command(args, {}) == "a1"
        assert mirror_out.called

    def test_routes_to_case_and_entries(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "get_incident_type", return_value=Elasticsearch_v2.INCIDENT_TYPE_CASE)
        mirror_case = mocker.patch.object(Elasticsearch_v2, "_mirror_out_case")
        mirror_entries = mocker.patch.object(Elasticsearch_v2, "_mirror_out_case_entries")
        args = {
            "remoteId": "c1",
            "data": {"type": Elasticsearch_v2.INCIDENT_TYPE_CASE},
            "entries": [{"ID": "1"}],
            "incidentChanged": True,
            "delta": {"title": "new"},
            "status": 1,
        }
        assert Elasticsearch_v2.update_remote_system_command(args, {}) == "c1"
        assert mirror_case.called
        assert mirror_entries.called


class TestMirrorOutCaseEntries:
    """Tests for _mirror_out_case_entries."""

    def test_posts_comment_for_tagged_entry(self, mocker):
        import Elasticsearch_v2

        req = mocker.patch.object(Elasticsearch_v2, "kibana_http_request")
        Elasticsearch_v2._mirror_out_case_entries("c1", [{"ID": "1", "Tags": ["comment"], "Contents": "hello"}], {})
        assert req.called
        _, kwargs = req.call_args
        assert kwargs["json_data"]["comment"] == "hello"

    def test_skips_untagged_entry(self, mocker):
        import Elasticsearch_v2

        req = mocker.patch.object(Elasticsearch_v2, "kibana_http_request")
        Elasticsearch_v2._mirror_out_case_entries("c1", [{"ID": "1", "Tags": [], "Contents": "hello"}], {})
        assert not req.called

    def test_serializes_dict_contents(self, mocker):
        import Elasticsearch_v2

        req = mocker.patch.object(Elasticsearch_v2, "kibana_http_request")
        Elasticsearch_v2._mirror_out_case_entries("c1", [{"ID": "1", "Tags": ["comment"], "Contents": {"a": 1}}], {})
        _, kwargs = req.call_args
        assert kwargs["json_data"]["comment"] == json.dumps({"a": 1})


class TestMirrorOutSecurityAlert:
    """Tests for _mirror_out_security_alert."""

    def test_updates_status(self, mocker):
        import Elasticsearch_v2

        req = mocker.patch.object(Elasticsearch_v2, "kibana_http_request", return_value={"updated": 1})
        Elasticsearch_v2._mirror_out_security_alert("a1", {"status": "acknowledged"}, 1, {})
        assert req.called
        _, kwargs = req.call_args
        assert kwargs["json_data"]["status"] == "acknowledged"

    def test_nothing_to_mirror_logs_error(self, mocker):
        import Elasticsearch_v2

        req = mocker.patch.object(Elasticsearch_v2, "kibana_http_request")
        err = mocker.patch.object(Elasticsearch_v2, "mirror_error")
        Elasticsearch_v2._mirror_out_security_alert("a1", {"severity": 3}, 1, {})
        assert not req.called
        assert err.called

    def test_close_sets_status_and_reason(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "CLOSE_ELASTIC_INCIDENT", True)
        req = mocker.patch.object(Elasticsearch_v2, "kibana_http_request", return_value={"updated": 1})
        Elasticsearch_v2._mirror_out_security_alert("a1", {"closeReason": "Resolved"}, Elasticsearch_v2.IncidentStatus.DONE, {})
        _, kwargs = req.call_args
        assert kwargs["json_data"]["status"] == "closed"
        assert kwargs["json_data"]["reason"] == "true_positive"

    def test_updates_tags(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(
            Elasticsearch_v2,
            "search_security_alerts",
            return_value={"hits": {"hits": [{"_id": "doc1", "_source": {"kibana.alert.workflow_tags": ["old"]}}]}},
        )
        req = mocker.patch.object(Elasticsearch_v2, "kibana_http_request")
        Elasticsearch_v2._mirror_out_security_alert("a1", {"tags": ["new"]}, 1, {})
        # Last call should be to the tags endpoint.
        args, kwargs = req.call_args
        assert args[1] == "/api/detection_engine/signals/tags"
        assert kwargs["json_data"]["tags"]["tags_to_add"] == ["new"]
        assert kwargs["json_data"]["tags"]["tags_to_remove"] == ["old"]


class TestMirrorOutCase:
    """Tests for _mirror_out_case."""

    def test_case_not_found_skips(self, mocker):
        import Elasticsearch_v2

        req = mocker.patch.object(Elasticsearch_v2, "kibana_http_request", return_value=None)
        Elasticsearch_v2._mirror_out_case("c1", {"title": "x"}, 1, {}, {})
        # Only the initial GET was attempted.
        assert req.call_count == 1

    def test_updates_supported_fields(self, mocker):
        import Elasticsearch_v2

        calls = []

        def fake_request(method, path, **kwargs):
            calls.append((method, path, kwargs))
            if method == "GET":
                return {"version": "v1"}
            return {}

        mocker.patch.object(Elasticsearch_v2, "kibana_http_request", side_effect=fake_request)
        Elasticsearch_v2._mirror_out_case("c1", {"title": "new title", "severity": 3}, 1, {}, {})
        patch_call = [c for c in calls if c[0] == "PATCH"][0]
        case_payload = patch_call[2]["json_data"]["cases"][0]
        assert case_payload["title"] == "new title"
        assert case_payload["severity"] == "high"

    def test_no_supported_fields_skips_patch(self, mocker):
        import Elasticsearch_v2

        calls = []

        def fake_request(method, path, **kwargs):
            calls.append(method)
            if method == "GET":
                return {"version": "v1"}
            return {}

        mocker.patch.object(Elasticsearch_v2, "kibana_http_request", side_effect=fake_request)
        Elasticsearch_v2._mirror_out_case("c1", {"unsupported": "x"}, 1, {}, {})
        assert "PATCH" not in calls


class TestGetModifiedRemoteData:
    """Tests for get_modified_remote_data_command and its helpers."""

    def test_get_modified_alert_ids_empty_update(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2._get_modified_alert_ids("", {}) == []

    def test_get_modified_alert_ids_uses_uuid_then_docid(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(
            Elasticsearch_v2,
            "search_security_alerts",
            return_value={
                "hits": {
                    "hits": [
                        {"_id": "doc1", "_source": {"kibana.alert.uuid": "uuid1"}},
                        {"_id": "doc2", "_source": {}},
                    ]
                }
            },
        )
        ids = Elasticsearch_v2._get_modified_alert_ids("2020-01-01T00:00:00Z", {})
        assert ids == ["uuid1", "doc2"]

    def test_get_modified_case_ids_filters_by_time(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(
            Elasticsearch_v2,
            "kibana_http_request",
            return_value={
                "cases": [
                    {"id": "new", "updated_at": "2020-01-02T00:00:00Z"},
                    {"id": "old", "updated_at": "2019-12-01T00:00:00Z"},
                ]
            },
        )
        ids = Elasticsearch_v2._get_modified_case_ids("2020-01-01T00:00:00Z", {})
        assert ids == ["new"]

    def test_get_modified_case_ids_bad_update(self):
        import Elasticsearch_v2

        assert Elasticsearch_v2._get_modified_case_ids("not-a-date", {}) == []

    def test_command_dedups_ids(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "_get_modified_alert_ids", return_value=["a", "b"])
        mocker.patch.object(Elasticsearch_v2, "_get_modified_case_ids", return_value=["b", "c"])
        response = Elasticsearch_v2.get_modified_remote_data_command({"lastUpdate": "2020-01-01T00:00:00Z"}, {})
        assert response.modified_incident_ids == ["a", "b", "c"]

    def test_command_rate_limit_on_alerts(self, mocker):
        import Elasticsearch_v2

        mocker.patch.object(Elasticsearch_v2, "_get_modified_alert_ids", side_effect=Exception("429"))
        return_error = mocker.patch.object(Elasticsearch_v2, "return_error", side_effect=SystemExit)
        with pytest.raises(SystemExit):
            Elasticsearch_v2.get_modified_remote_data_command({"lastUpdate": "2020-01-01T00:00:00Z"}, {})
        assert return_error.called