from collections import namedtuple from copy import deepcopy from unittest.mock import MagicMock, patch import demistomock as demisto import pytest import SplunkPyV2 as splunk from CommonServerPython import * from pytest_mock import MockerFixture from splunklib import client, results from splunklib.binding import AuthenticationError, HTTPError from datetime import UTC RETURN_ERROR_TARGET = "SplunkPyV2.return_error" DICT_RAW_RESPONSE = ( '"1528755951, url="https://test.url.com", search_name="NG_SIEM_UC25- High number of hits against ' 'unknown website from same subnet", action="allowed", dest="bb.bbb.bb.bbb , cc.ccc.ccc.cc , ' 'xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", distinct_hosts="5", ' 'first_3_octets="1.1.1", first_time="06/11/18 17:34:07 , 06/11/18 17:37:55 , 06/11/18 17:41:28 , ' '06/11/18 17:42:05 , 06/11/18 17:42:38", info_max_time="+Infinity", info_min_time="0.000", ' 'src="xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", u_category="unknown", ' 'user="xyz\\a1234 , xyz\\b5678 , xyz\\c91011 , xyz\\d121314 , unknown", website="2.2.2.2""' ) DICT_RAW_RESPONSE_WITH_MESSAGE_ID = ( '"1528755951, message-id="1", url="https://test.url.com", ' 'search_name="NG_SIEM_UC25- High number of hits against ' 'unknown website from same subnet", action="allowed", dest="bb.bbb.bb.bbb , ' 'cc.ccc.ccc.cc , xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", ' 'distinct_hosts="5", ' 'first_3_octets="1.1.1", first_time="06/11/18 17:34:07 , ' "06/11/18 17:37:55 , 06/11/18 17:41:28 , " '06/11/18 17:42:05 , 06/11/18 17:42:38", info_max_time="+Infinity", info_min_time="0.000", ' 'src="xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", u_category="unknown", ' 'user="xyz\\a1234 , xyz\\b5678 , xyz\\c91011 , xyz\\d121314 , unknown", website="2.2.2.2""' ) LIST_RAW = ( "Feb 13 09:02:55 1,2020/02/13 09:02:55,001606001116,THREAT,url," "1,2020/02/13 09:02:55,10.1.1.1,1.2.3.4,0.0.0.0,0.0.0.0,rule1,jordy,,web-browsing,vsys1,trust,untrust," "ethernet1/2,ethernet1/1,forwardAll,2020/02/13 09:02:55,59460,1,62889,80,0,0,0x208000,tcp,alert," '"ushship.com/xed/config.bin",(9999),not-resolved,informational,client-to-server,' "0,0x0,1.1.22.22-5.6.7.8,United States,0,text/html" ) RAW_WITH_MESSAGE = ( '{"@timestamp":"2019-10-15T13:30:08.578-04:00","message":"{"TimeStamp":"2019-10-15 13:30:08",' '"CATEGORY_1":"CONTACT","ASSOCIATEOID":"G2N2TJETBRAAX68V","HOST":' '"step-up-authentication-api.gslb.es.oneadp.com","SCOPE[4]":"PiSvcsProvider\\/payroll","SCOPE[19]":' '"\\/api\\/events\\/core\\/v1\\/user-status","CONTEXT":"\\/smsstepup","FLOW":"API","X-REAL-IP":' '"2.2.2.2","PRODUCT_CODE":"WFNPortal","X-FORWARDED-PROTO":"http","ERROR_ID":"4008",' '"SCOPE[23]":"\\/security\\/notification-communication-response-value.accept","REQ_URL":' '"http:\\/\\/step-up-authentication-api.gslb.es.blabla.com\\/smsstepup\\/events\\/core\\/v1\\/step-up-' 'user-authorization-request.evaluate","SCOPE[35]":"autopay\\/payroll\\/v1\\/cafeteria-plan-' 'configurations\\/{configurationItemID}","SCOPE_MATCHED":"Y","SCOPE[43]":"communication\\/n' 'otification-message-template.add","SCOPE[11]":"\\/ISIJWSUserSecurity","SCOPE[27]":"autopay\\/events' '\\/payroll\\/v1\\/earning-configuration.add","ORGOID":"G2SY6MR3ATKA232T","SCOPE[8]":"\\/' 'ISIJWSAssociatesService","SCOPE[39]":"autopay\\/payroll\\/v1\\/earning-configurations",' '"SETUP_SELF":"N","SCOPE[47]":"communication\\/notification.publish","SCOPE[15]":"' '\\/OrganizationSoftPurge","X-FORWARDED-HOST":"step-up-authentication-api.gslb.es.blabla.com",' '"ADP-MESSAGEID":"a1d57ed2-1fe6-4800-be7a-26cd89bhello","CNAME":"JRJG INC","CONTENT-LENGTH":' '"584","SCOPE[31]":"autopay\\/events\\/payroll\\/v1\\/earning-configuration.remove","CID":"BSTAR00044"' ',"ACTOR_UID":"ABinters@BSTAR00044","SECURE_API_MODE":"HTTPS_SECURE","X-REQUEST-ID":' '"2473a981bef27bc8444e510adc12234a","SCOPE[1]":"AVSSCP\\/Docstash\\/Download","SCOPE[18]":' '"\\/api\\/events\\/core\\/v1\\/product-role.assign","BLOCK_SESSION":"Y","CONSUMER_ID":' '"ab2e715e-41c4-43d6-bff7-fc2d713hello","SCOPE[34]":"autopay\\/payroll\\/v1\\/cafeteria-plan-' 'configurations","SCOPE[46]":"communication\\/notification-message-template.remove","MODULE":' '"STEPUP_API","SCOPE[9]":"\\/ISIJWSClientService","SCOPE[10]":"\\/ISIJWSJobsService","SCOPE[22]":' '"\\/api\\/person-account-registration","SCOPE[38]":"autopay\\/payroll\\/v1\\/deposit-configurations",' '"SUBJECT_ORGOID":"G2SY6MR3ATKA232T","SCOPE[5]":"\\/Associate","SCOPE[14]":"\\/Organization",' '"SCOPE[26]":"WFNSvcsProvider\\/payrollPi","EVENT_ID":"9ea87118-5679-5b0e-a67f-1abd8ccabcde",' '"SCOPE[30]":"autopay\\/events\\/payroll\\/v1\\/earning-configuration.payroll-accumulators.modify",' '"X-FORWARDED-PORT":"80","SCOPE[42]":"autopay\\/payroll\\/v1\\/worker-employment-records","JTI":' '"867b6d06-47cf-40ab-8dd7-bd0d57babcde","X-DOMAIN":"secure.api.es.abc.com","SOR_CODE":' '"WFNPortal","SCOPE[29]":"autopay\\/events\\/payroll\\/v1\\/earning-configuration.configuration' '-tags.modify","SCOPE[2]":"AVSSCP\\/Docstash\\/Get","OUTPUT_TYPE":"FAIL","ERR_MSG":"BLOCK_SESSION",' '"TRANS_ID":"3AF-D30-7CTTCQ","SCOPE[45]":"communication\\/notification-message-template.read",' '"USE_HISTORY":"Y","SCHEME":"http","SCOPE[13]":"\\/ISIJWSUsersService","SCOPE[21]":"\\/api\\/person",' '"SCOPE[33]":"autopay\\/events\\/payroll\\/v1\\/worker-insurable-payments.modify","X-FORWARDED-FOR":' '"8.8.8.8, 10.10.10.10, 1.2.3.4, 5.6.7.8","SCOPE[17]":"\\/api\\/core\\/v1\\/organization",' '"SCOPE[25]":"\\/step-up-user-authorization.initiate","SCOPE[6]":"\\/Associate\\/PIC","SCOPE[37]":' '"autopay\\/payroll\\/v1\\/cafeteria-plan-configurations\\/{configurationItemID}\\/' 'payroll-item-configurations\\/{payrollItemID}","FLOW_TYPE":"REST","SCOPE[41]":' '"autopay\\/payroll\\/v1\\/payroll-output","CONSUMERAPPOID":"WFNPortal","RESOURCE":' '"\\/events\\/core\\/v1\\/step-up-user-authorization-request.evaluate","USER-AGENT":' '"Apache-HttpClient\\/4.5.5 (Java\\/10.0.1)","SCOPE[3]":"AVSSCP\\/Docstash\\/List",' '"SUB_CATEGORY_1":"worker.businessCommunication.email.change","TIME":"9","X-SCHEME":' '"http","ADP-CONVERSATIONID":"stY46PpweABoT5JX04CZGCeBbX8=","SCOPE[12]":' '"\\/ISIJWSUserSecurityService","SCOPE[24]":"\\/step-up-user-authorization-request.evaluate",' '"SCOPE[32]":"autopay\\/events\\/payroll\\/v1\\/retro-pay-request.add","SCOPE[44]":' '"communication\\/notification-message-template.change","ACTION":"POST","SCOPE[7]":' '"\\/AssociateSoftPurge","SCOPE[16]":"\\/api\\/authentication","X-ORIGINAL-URI":' '"\\/smsstepup\\/events\\/core\\/v1\\/step-up-user-authorization-request.evaluate","SCOPE[28]":' '"autopay\\/events\\/payroll\\/v1\\/earning-configuration.change","SCOPE[36]":' '"autopay\\/payroll\\/v1\\/cafeteria-plan-configurations\\/{configurationItemID}\\/payroll-item' '-configurations","SESSION_ID":"f50be909-9e4f-408d-bf77-68499012bc35","SCOPE[20]":' '"\\/api\\/events\\/core\\/v1\\/user.provision","SUBJECT_AOID":"G370XX6XYCABCDE",' '"X-ORIGINAL-FORWARDED-FOR":"1.1.1.1, 3.3.3.3, 4.4.4.4","SCOPE[40]":' '"autopay\\/payroll\\/v1\\/employer-details"}","TXID":"3AF-D30-ABCDEF","ADP-MessageID":' '"a1d57ed2-1fe6-4800-be7a-26cd89bf686d","SESSIONID":"stY46PpweFToT5JX04CZGMeCvP8=","ORGOID":' '"G2SY6MR3ATKA232T","AOID":"G2N2TJETBRAAXAAA","MSGID":"a1d57ed2-1fe6-0000-be7a-26cd89bf686d"}' ) SAMPLE_RESPONSE = [ results.Message("INFO-TEST", "test message"), { "_bkt": "finding~668~66D21DF4-F4FD-4886-A986-82E72ADCBFE9", "_cd": "668:17198", "_indextime": "1596545116", "_raw": '1596545116, search_name="Endpoint - Recurring Malware Infection - Rule", count="17", ' 'day_count="8", dest="ACME-workstation-012", info_max_time="1596545100.000000000", ' 'info_min_time="1595939700.000000000", info_search_time="1596545113.965466000", ' 'signature="Trojan.Gen.2"', "_serial": "50", "_si": ["ip-172-31-44-193", "finding"], "_sourcetype": "stash", "_time": "2020-08-04T05:45:16.000-07:00", "dest": "ACME-workstation-012", "dest_asset_id": "028877d3c80cb9d87900eb4f9c9601ea993d9b63", "dest_asset_tag": ["cardholder", "pci", "americas"], "dest_bunit": "americas", "dest_category": ["cardholder", "pci"], "dest_city": "Pleasanton", "dest_country": "USA", "dest_ip": "192.168.3.12", "dest_is_expected": "TRUE", "dest_lat": "37.694452", "dest_long": "-121.894461", "dest_nt_host": "ACME-workstation-012", "dest_pci_domain": ["trust", "cardholder"], "dest_priority": "medium", "dest_requires_av": "TRUE", "dest_risk_object_type": "system", "dest_risk_score": "15680", "dest_should_timesync": "TRUE", "dest_should_update": "TRUE", "host": "ip-172-31-44-193", "host_risk_object_type": "system", "host_risk_score": "0", "index": "finding", "linecount": "1", "priorities": "medium", "priority": "medium", "risk_score": "15680", "rule_description": "Endpoint - Recurring Malware Infection - Rule", "rule_name": "Endpoint - Recurring Malware Infection - Rule", "rule_title": "Endpoint - Recurring Malware Infection - Rule", "security_domain": "Endpoint - Recurring Malware Infection - Rule", "severity": "unknown", "signature": "Trojan.Gen.2", "source": "Endpoint - Recurring Malware Infection - Rule", "sourcetype": "stash", "splunk_server": "ip-172-31-44-193", "urgency": "low", "owner": "unassigned", "event_id": "66D21DF4-F4FD-4886-A986-82E72ADCBFE9@@finding@@5aa44496ec8e5cf45c78ab230189a4ca", }, { "_bkt": "finding~3252~66D21DF4-F4FD-4886-A986-82E72ADCBFE9", "_cd": "3252:4913", "_eventtype_color": "none", "_indextime": "1737544322", "_raw": '1596545116, search_name="Endpoint - Recurring Malware Infection - Rule", count="17", ' 'day_count="8", dest="ACME-workstation-012", info_max_time="1596545100.000000000", ' 'info_min_time="1595939700.000000000", info_search_time="1596545113.965466000", ' 'signature="Trojan.Gen.2"', "_serial": "12", "_si": ["ip-1-1-1-1", "finding"], "_sourcetype": "stash", "_time": "2025-01-22T11:12:02.000+00:00", "comment": [ "change all fields", "changed to in progress", "changed to pending", ], "count": "1", "drilldown_earliest": "0.000", "drilldown_earliest_offset": "0.000", "drilldown_latest": "+Infinity", "drilldown_latest_offset": "+Infinity", "drilldown_name": "View infections on ACME-code-001", "drilldown_search": '| from datamodel:"Malware"."Malware_Attacks" | search dest="ACME-code-001"', "event_hash": "734b6c7bcd700ccd0449575164772230", "event_id": "test_id", "eventtype": "modfinding_results finding modfinding_results finding", "extract_assets": '["src", "dest", "dvc", "orig_host"]', "extract_identities": '["src_user", "user"]', "host": "ip-172-31-44-193", "host_risk_object_type": "system", "host_risk_score": "0", "index": "finding", "indexer_guid": "66D21DF4-F4FD-4886-A986-82E72ADCBFE9", "info_max_time": "+Infinity", "info_min_time": "0.000", "info_search_time": "1737504174.093091000", "investigation_profiles": "{}", "lastTime": "1737544072", "linecount": "2", "orig_action_name": "finding", "orig_rid": "0.6979", "owner": "test_owner", "owner_realname": "test_owner", "priorities": "critical", "priority": "critical", "review_time": "1737547610.488234", "reviewer": ["test_owner", "test_owner", "test_owner"], "risk_score": "24160", "rule_description": "A high or critical priority host (ACME-code-001) was detected with malware.", "rule_id": "test_id", "rule_name": "High Or Critical Priority Host With Malware Detected", "rule_title": "High Or Critical Priority Host With Malware Detected", "savedsearch_description": "Alerts when an infection is noted on a host with high or critical priority.", "search_name": "Endpoint - High Or Critical Priority Host With Malware - Rule", "security_domain": "endpoint", "severity": "high", "signature": "127", "source": "Endpoint - High Or Critical Priority Host With Malware - Rule", "sourcetype": "stash", "splunk_server": "ip-1-1-1-1", "status": "3", "status_default": "false", "status_description": "Closure is pending some action.", "status_end": "false", "status_group": "Open", "status_label": "Pending", "tag": ["modaction_result", "test_user"], "tag::eventtype": "modaction_result", "timestamp": "none", "urgency": "informational", }, ] SAMPLE_AUDIT_INDEX_RESPONSE = [ { "_key": "test_id_1737547610.49", "review_time": "1737547610.488234", "owner": "test_owner", "owner_realname": "test_owner", "reviewer": "test_owner", "reviewer_realname": "test_owner", "rule_id": "test_id", "rule_name": "High Or Critical Priority Host With Malware Detected", "status": "3", "status_default": "false", "status_description": "Closure is pending some action.", "status_end": "false", "status_group": "Open", "status_label": "Pending", "urgency": "informational", } ] EXPECTED = { "action": "allowed", "dest": "bb.bbb.bb.bbb , cc.ccc.ccc.cc , xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", "distinct_hosts": "5", "first_3_octets": "1.1.1", "first_time": "06/11/18 17:34:07 , 06/11/18 17:37:55 , 06/11/18 17:41:28 , 06/11/18 17:42:05 , 06/11/18 17:42:38", "info_max_time": "+Infinity", "info_min_time": "0.000", "search_name": "NG_SIEM_UC25- High number of hits against unknown website from same subnet", "src": "xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", "u_category": "unknown", "user": "xyz\\a1234 , xyz\\b5678 , xyz\\c91011 , xyz\\d121314 , unknown", "website": "2.2.2.2", "url": "https://test.url.com", } EXPECTED_WITH_MESSAGE_ID = { "message-id": "1", "action": "allowed", "dest": "bb.bbb.bb.bbb , cc.ccc.ccc.cc , xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", "distinct_hosts": "5", "first_3_octets": "1.1.1", "first_time": "06/11/18 17:34:07 , 06/11/18 17:37:55 , 06/11/18 17:41:28 , 06/11/18 17:42:05 , 06/11/18 17:42:38", "info_max_time": "+Infinity", "info_min_time": "0.000", "search_name": "NG_SIEM_UC25- High number of hits against unknown website from same subnet", "src": "xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", "u_category": "unknown", "user": "xyz\\a1234 , xyz\\b5678 , xyz\\c91011 , xyz\\d121314 , unknown", "website": "2.2.2.2", "url": "https://test.url.com", } URL_TESTING_IN = '"url="https://test.com?key=val"' URL_TESTING_OUT = {"url": "https://test.com?key=val"} # checking a case where the last character for each value was cut RESPONSE = ( "NAS-IP-Address=2.2.2.2, NAS-Port=50222, NAS-Identifier=de-wilm-251littl-idf3b-s2, NAS-Port-Type=" "Ethernet, NAS-Port-Id=GigabitEthernet2/0/05" ) POSITIVE = { "NAS-IP-Address": "2.2.2.2", "NAS-Identifier": "de-wilm-251littl-idf3b-s2", "NAS-Port": "50222", "NAS-Port-Id": "GigabitEthernet2/0/05", "NAS-Port-Type": "Ethernet", } # testing the ValueError and json sections RAW_JSON = '{"Test": "success"}' RAW_STANDARD = '"Test="success"' RAW_JSON_AND_STANDARD_OUTPUT = {"Test": "success"} class Jobs: def __init__(self, status, service): self.oneshot = lambda x, **kwargs: x state = namedtuple("state", "content") self.state = state(content={"dispatchState": str(status)}) self.service = service def __getitem__(self, arg): return 0 def create(self, query, **kwargs): job = client.Job(sid="123456", service=self.service, **kwargs) job.resultCount = 0 job._state = self.state return job class _MagicKVStore: """Minimal stand-in for ``service.kvstore`` used by tests. The production code reaches the KV-store via ``service.kvstore["mc_notes"].data.query(query=...)`` (see :func:`SplunkPyV2.enrich_with_splunk_notes_v2`). The test ``Service`` mock historically did not expose ``kvstore``, which made the v2 helper raise an ``AttributeError`` and spill a ``demisto.error(...)`` line into stdout — flagged by the strict autouse ``check_std_out_err`` fixture in :mod:`conftest`. This stand-in returns a benign empty result for any collection access so the helper short-circuits via the "no notes found" branch without touching real Splunk infrastructure. Tests that need to assert specific KV-store behavior can still override ``service.kvstore`` (or the higher-level helper) with their own mock — this default is only the safety net. """ def __getitem__(self, _name): # ``service.kvstore["mc_notes"]`` return MagicMock(data=MagicMock(query=MagicMock(return_value=[]))) def __contains__(self, _name): # ``"mc_notes" in service.kvstore`` return True class Service: def __init__(self, status): self.jobs = Jobs(status, self) self.status = status self.disable_v2_api = False self.namespace = {"app": "test", "owner": "test", "sharing": "global"} self._abspath = lambda x, **kwargs: x # See :class:`_MagicKVStore` docstring for rationale. self.kvstore = _MagicKVStore() def get(self, path_segment, owner=None, app=None, headers=None, sharing=None, **query): return {"status": "200", "body": "test", "headers": {"content-type": "application/json"}, "reason": "OK"} def job(self, sid): return self.jobs def test_raw_to_dict(): actual_raw = DICT_RAW_RESPONSE response = splunk.raw_to_dict(actual_raw) response_with_message = splunk.raw_to_dict(DICT_RAW_RESPONSE_WITH_MESSAGE_ID) list_response = splunk.raw_to_dict(LIST_RAW) raw_message = splunk.raw_to_dict(RAW_WITH_MESSAGE) empty = splunk.raw_to_dict("") url_test = splunk.raw_to_dict(URL_TESTING_IN) character_check = splunk.raw_to_dict(RESPONSE) assert response == EXPECTED assert response_with_message == EXPECTED_WITH_MESSAGE_ID assert list_response == {} assert raw_message.get("SCOPE[29]") == "autopay\\/events\\/payroll\\/v1\\/earning-configuration.configuration-tags.modify" assert isinstance(raw_message, dict) assert empty == {} assert url_test == URL_TESTING_OUT assert character_check == POSITIVE assert splunk.raw_to_dict(RAW_JSON) == RAW_JSON_AND_STANDARD_OUTPUT assert splunk.raw_to_dict(RAW_STANDARD) == RAW_JSON_AND_STANDARD_OUTPUT assert splunk.raw_to_dict('drilldown_search="key IN ("test1","test2")') == {"drilldown_search": "key IN (test1,test2)"} assert splunk.raw_to_dict( '123456, sample_account="sample1", sample_account="sample2", sample_account="sample3", distinct_count_ac="5"' ) == {"sample_account": "sample1, sample2, sample3", "distinct_count_ac": "5"} @pytest.mark.parametrize( "text, output", [ ("", [""]), ('"",', ['"",']), # a value shouldn't do anything special ("woopwoop", ["woopwoop"]), # a normal key value without quotes ("abc=123", ['abc="123"']), # add a comma at the end ("abc=123,", ['abc="123"']), # a normal key value with quotes ('cbd="123"', ['cbd="123"']), # check all wrapped with quotes removed ('"abc="123""', ['abc="123"']), # we need to remove 111 at the start. ('111, cbd="123"', ['cbd="123"']), # Testing with/without quotes and/or spaces: ("abc=123,cbd=123", ['abc="123"', 'cbd="123"']), ('abc=123,cbd="123"', ['abc="123"', 'cbd="123"']), ('abc="123",cbd=123', ['abc="123"', 'cbd="123"']), ('abc="123",cbd="123"', ['abc="123"', 'cbd="123"']), ("abc=123, cbd=123", ['abc="123"', 'cbd="123"']), ('abc=123, cbd="123"', ['abc="123"', 'cbd="123"']), ('cbd="123", abc=123', ['abc="123"', 'cbd="123"']), ('cbd="123",abc=123', ['abc="123"', 'cbd="123"']), # Continue testing quotes with more values: ("xyz=321,cbd=123,abc=123", ['xyz="321"', 'abc="123"', 'cbd="123"']), ('xyz=321,cbd="123",abc=123', ['xyz="321"', 'abc="123"', 'cbd="123"']), ('xyz="321",cbd="123",abc=123', ['xyz="321"', 'abc="123"', 'cbd="123"']), ('xyz="321",cbd="123",abc="123"', ['xyz="321"', 'abc="123"', 'cbd="123"']), # Testing nested quotes (the main reason for quote_group): # Try to remove the start 111. ('111, cbd="a="123""', ['cbd="a="123""']), ('cbd="a="123""', ['cbd="a="123""']), ('cbd="a="123", b=321"', ['cbd="a="123", b="321""']), ('cbd="a=123, b=321"', ['cbd="a="123", b="321""']), ('cbd="a=123, b="321""', ['cbd="a="123", b="321""']), ('cbd="a="123", b="321""', ['cbd="a="123", b="321""']), ('cbd="a=123, b=321"', ['cbd="a="123", b="321""']), ('xyz=123, cbd="a="123", b=321"', ['xyz="123"', 'cbd="a="123", b="321""']), ('xyz="123", cbd="a="123", b="321""', ['xyz="123"', 'cbd="a="123", b="321""']), ('xyz="123", cbd="a="123", b="321"", qqq=2', ['xyz="123"', 'cbd="a="123", b="321""', 'qqq="2"']), ('xyz="123", cbd="a="123", b="321"", qqq="2"', ['xyz="123"', 'cbd="a="123", b="321""', 'qqq="2"']), ], ) def test_quote_group(text, output): assert sorted(splunk.quote_group(text)) == sorted(output) data_test_replace_keys = [ ({}, {}), ({"test": "test"}, {"test": "test"}), ({"test.": "test."}, {"test_": "test."}), ({"te.st": "te.st"}, {"te_st": "te.st"}), ({"te[st": "te[st"}, {"te_st": "te[st"}), ({"te]st": "te]st"}, {"te_st": "te]st"}), ({"te)st": "te)st"}, {"te_st": "te)st"}), ({"te(st": "te(st"}, {"te_st": "te(st"}), ("", ""), (None, None), ] @pytest.mark.parametrize("dict_in, dict_out", data_test_replace_keys) def test_replace_keys(dict_in, dict_out): out = splunk.replace_keys(deepcopy(dict_in)) assert out == dict_out, f"replace_keys({dict_in}) got: {out} instead: {dict_out}" def test_splunk_submit_event_hec_command(mocker): text = "a msg with a failure." class MockRes: def __init__(self, text): self.text = text mocker.patch.object(splunk, "splunk_submit_event_hec", return_value=MockRes(text)) return_error_mock = mocker.patch(RETURN_ERROR_TARGET) splunk.splunk_submit_event_hec_command(params={"hec_url": "mock_url"}, args={"entry_id": "some_entry"}, service=Service) err_msg = return_error_mock.call_args[0][0] assert err_msg == f"Could not send event to Splunk {text}" def check_request_channel(args: dict): """ Check if args contains a request_channel, return the proper text. Args: args: A dict of args. Returns: A MockResRequestChannel with the correct text value. """ if args.get("request_channel"): return MockResRequestChannel('{"text":"Success","code":0,"ackId":1}') else: return MockResRequestChannel('{"text":"Data channel is missing","code":10}') class MockResRequestChannel: def __init__(self, text): self.text = text def test_splunk_submit_event_hec_command_request_channel(mocker): """ Given - An args dict that contains a request_channel and a dummy params. When - Executing splunk_submit_event_hec_command function Then - The return result object contains the correct message. """ args = {"request_channel": "11111111-1111-1111-1111-111111111111", "entry_id": "some_entry"} mocker.patch.object(splunk, "splunk_submit_event_hec", return_value=check_request_channel(args)) moc = mocker.patch.object(demisto, "results") splunk.splunk_submit_event_hec_command(params={"hec_url": "mock_url"}, args=args, service=Service) readable_output = moc.call_args[0][0] assert readable_output == "The events were sent successfully to Splunk. AckID: 1" def test_splunk_submit_event_hec_command_without_request_channel(mocker): """ Given - An args dict that doesn't contain a request_channel and a dummy params. When - Executing splunk_submit_event_hec_command function Then - The return result object contains the correct message. """ args = {"entry_id": "some_entry"} mocker.patch.object(splunk, "splunk_submit_event_hec", return_value=check_request_channel(args)) return_error_mock = mocker.patch(RETURN_ERROR_TARGET) splunk.splunk_submit_event_hec_command(params={"hec_url": "mock_url"}, args=args, service=Service) err_msg = return_error_mock.call_args[0][0] assert err_msg == 'Could not send event to Splunk {"text":"Data channel is missing","code":10}' SEARCH_RESULT = [ {"But": {"This": "is"}, "Very": "Unique"}, {"Something": "regular", "But": {"This": "is"}, "Very": "Unique"}, {"Something": "natural", "But": {"This": "is a very very"}, "Very": "Unique and awesome"}, ] REGULAR_ALL_CHOSEN_FIELDS = ["Something", "But", "Very"] REGULAR_CHOSEN_FIELDS_SUBSET = ["Something", "Very"] REGEX_CHOSEN_FIELDS_SUBSET = ["Some*", "Very"] NON_EXISTING_FIELDS = ["SDFAFSD", "ASBLFKDJK"] @pytest.mark.parametrize( "search_result, chosen_fields, expected_result", [ (SEARCH_RESULT, REGULAR_ALL_CHOSEN_FIELDS, REGULAR_ALL_CHOSEN_FIELDS), (SEARCH_RESULT, REGULAR_CHOSEN_FIELDS_SUBSET, REGULAR_CHOSEN_FIELDS_SUBSET), (SEARCH_RESULT, REGEX_CHOSEN_FIELDS_SUBSET, REGULAR_CHOSEN_FIELDS_SUBSET), (SEARCH_RESULT, NON_EXISTING_FIELDS, []), ], ) def test_commands(search_result, chosen_fields, expected_result): from SplunkPyV2 import update_headers_from_field_names headers = update_headers_from_field_names(search_result, chosen_fields) assert expected_result == headers APPS = ["app"] STORES = ["store"] EMPTY_CASE = {} STORE_WITHOUT_APP = {"kv_store_collection_name": "test"} JUST_APP_NAME = {"app_name": "app"} # happens in splunk-kv-store-collections-list command CREATE_COMMAND = {"app_name": "app", "kv_store_name": "not_store"} CORRECT = {"app_name": "app", "kv_store_collection_name": "store"} INCORRECT_STORE_NAME = {"app_name": "app", "kv_store_collection_name": "not_store"} data_test_check_error = [ (EMPTY_CASE, "app not found"), (STORE_WITHOUT_APP, "app not found"), (JUST_APP_NAME, "empty"), (CREATE_COMMAND, "empty"), (CORRECT, "empty"), (INCORRECT_STORE_NAME, "KV Store not found"), ] @pytest.mark.parametrize("args, out_error", data_test_check_error) def test_check_error(args, out_error): class Service: def __init__(self): self.apps = APPS self.kvstore = STORES try: splunk.check_error(Service(), args) raise splunk.DemistoException("empty") except splunk.DemistoException as error: output = str(error) assert output == out_error, f"check_error(service, {args})\n\treturns: {output}\n\tinstead: {out_error}" EMPTY_CASE = {} JUST_KEY = {"key": "key"} WITH_ALL_PARAMS = {"key": "demisto", "value": "is awesome", "limit": 1, "query": "test"} STANDARD_KEY_VAL = {"key": "demisto", "value": "is awesome"} KEY_AND_LIMIT = {"key": "key", "limit": 1} KEY_AND_QUERY = {"key": "key", "query": "test_query"} QUERY = {"query": "test_query"} QUERY_AND_VALUE = {"query": "test_query", "value": "awesome"} data_test_build_kv_store_query = [ (EMPTY_CASE, str(EMPTY_CASE)), (JUST_KEY, str(EMPTY_CASE)), (STANDARD_KEY_VAL, '{"demisto": "is awesome"}'), (WITH_ALL_PARAMS, '{"demisto": "is awesome"}'), (KEY_AND_LIMIT, {"limit": 1}), (KEY_AND_QUERY, "test_query"), (QUERY, "test_query"), (QUERY_AND_VALUE, "test_query"), ] @pytest.mark.parametrize("args, expected_query", data_test_build_kv_store_query) def test_build_kv_store_query(args, expected_query, mocker): mocker.patch("SplunkPyV2.get_key_type", return_value=None) output = splunk.build_kv_store_query(None, args) assert output == expected_query, f"build_kv_store_query({args})\n\treturns: {output}\n\tinstead: {expected_query}" data_test_build_kv_store_query_with_key_val = [ ({"key": "demisto", "value": "is awesome"}, str, '{"demisto": "is awesome"}'), ({"key": "demisto", "value": "1"}, int, '{"demisto": 1}'), ({"key": "demisto", "value": "True"}, bool, '{"demisto": true}'), ] @pytest.mark.parametrize("args, _type, expected_query", data_test_build_kv_store_query_with_key_val) def test_build_kv_store_query_with_key_val(args, _type, expected_query, mocker): mocker.patch("SplunkPyV2.get_key_type", return_value=_type) output = splunk.build_kv_store_query(None, args) assert output == expected_query, f"build_kv_store_query({args})\n\treturns: {output}\n\tinstead: {expected_query}" test_test_get_key_type = [ ({"field.key": "number"}, float), ({"field.key": "string"}, str), ({"field.key": "cidr"}, str), ({"field.key": "boolean"}, bool), ({"field.key": "empty"}, None), ({"field.key": "time"}, str), ] @pytest.mark.parametrize("keys_and_types, expected_type", test_test_get_key_type) def test_get_key_type(keys_and_types, expected_type, mocker): mocker.patch("SplunkPyV2.get_keys_and_types", return_value=keys_and_types) output = splunk.get_key_type(None, "key") assert output == expected_type, f"get_key_type(kv_store, key)\n\treturns: {output}\n\tinstead: {expected_type}" EMPTY_CASE = {} WITHOUT_FIELD = {"empty": "number"} STRING_FIELD = {"field.test": "string"} NUMBER_FIELD = {"field.test": "number"} INDEX = {"index.test": "string"} MIXED = {"field.test": "string", "empty": "field"} data_test_get_keys_and_types = [ (EMPTY_CASE, EMPTY_CASE), (WITHOUT_FIELD, EMPTY_CASE), (STRING_FIELD, {"field.test": "string"}), (NUMBER_FIELD, {"field.test": "number"}), (INDEX, {"index.test": "string"}), (MIXED, {"field.test": "string"}), ] @pytest.mark.parametrize("raw_keys, expected_keys", data_test_get_keys_and_types) def test_get_keys_and_types(raw_keys, expected_keys): class KVMock: def __init__(self): pass def content(self): return raw_keys output = splunk.get_keys_and_types(KVMock()) assert output == expected_keys, f"get_keys_and_types(kv_store)\n\treturns: {output}\n\tinstead: {expected_keys}" START_OUTPUT = "#### configuration for name store\n| field name | type |\n| --- | --- |" EMPTY_OUTPUT = "" STANDARD_CASE = {"field.test": "number"} STANDARD_OUTPUT = "\n| field.test | number |" data_test_get_kv_store_config = [({}, EMPTY_OUTPUT), (STANDARD_CASE, STANDARD_OUTPUT)] @pytest.mark.parametrize("fields, expected_output", data_test_get_kv_store_config) def test_get_kv_store_config(fields, expected_output, mocker): class Name: def __init__(self): self.name = "name" mocker.patch("SplunkPyV2.get_keys_and_types", return_value=fields) output = splunk.get_kv_store_config(Name()) expected_output = f"{START_OUTPUT}{expected_output}" assert output == expected_output class TestFetchRemovingIrrelevantIncidents: finding1 = {"status": "5", "event_id": "3"} finding2 = {"status": "6", "event_id": "4"} # In order to mock the service.jobs.oneshot() call in the fetch_findings function, we need to create # the following two classes class Jobs: def __init__(self): self.oneshot = lambda x, **kwargs: TestFetchForLateIndexedEvents.finding1 class Service: def __init__(self): self.jobs = TestFetchForLateIndexedEvents.Jobs() # Stand-in for service.kvstore — see _MagicKVStore docstring at module top. self.kvstore = _MagicKVStore() def test_remove_irrelevant_fetched_incident_ids(self, mocker: MockerFixture): """ Given - Incident IDs that were fetched in the last fetch round When - Fetching findings Then - Make sure that the fetched IDs that are no longer in the fetch window are removed """ from SplunkPyV2 import UserMappingObject mocker.patch.object(splunk, "get_current_splunk_time", return_value="2024-02-19T10:00:00.000000+0000") mocker.patch.object(demisto, "setLastRun") mock_last_run = { "next_run_earliest_time": "2024-02-12T10:00:00.000000+0000", "late_indexed_pagination": False, "next_run_found_incidents_ids": { "1": {"occurred_time": "2024-02-12T09:59:59.000000+0000"}, "2": {"occurred_time": "2024-02-18T10:00:00.000000+0000"}, }, } mock_params = {"fetchQuery": "`notable` is cool", "max_fetch": 2} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=[self.finding1, self.finding2]) service = self.Service() set_last_run_mocker = mocker.patch("demistomock.setLastRun") mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper) last_fetched_ids = set_last_run_mocker.call_args_list[0][0][0]["next_run_found_incidents_ids"] assert last_fetched_ids == { "2": {"occurred_time": "2024-02-18T10:00:00.000000+0000"}, "3": {"occurred_time": "2024-02-19T10:00:00.000000+0000"}, "4": {"occurred_time": "2024-02-19T10:00:00.000000+0000"}, } class TestFetchForLateIndexedEvents: finding1 = {"status": "5", "event_id": "id_1"} finding2 = {"status": "6", "event_id": "id_2"} # In order to mock the service.jobs.oneshot() call in the fetch_findings function, we need to create # the following two classes class Jobs: def __init__(self): self.oneshot = lambda x, **kwargs: TestFetchForLateIndexedEvents.finding1 class Service: def __init__(self): self.jobs = TestFetchForLateIndexedEvents.Jobs() # Stand-in for service.kvstore — see _MagicKVStore docstring at module top. self.kvstore = _MagicKVStore() # If late_indexed_pagination is True, then we exclude the last fetched ids (check by using fetch query), # and kwargs_oneshot['offset'] == 0 def test_fetch_query_and_oneshot_args(self, mocker: MockerFixture): """ Given - Mocked incidents api response - The key "late_indexed_pagination" in the last run object is set to True - Some incident IDs that were fetched in the last fetch round When - Fetching findings Then - Make sure that last fetched incident IDs are specified to be excluded from the fetch query - Make sure that the offset of the fetch query is set to 0 """ from SplunkPyV2 import UserMappingObject mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mocker.patch.object(demisto, "setLastRun") mock_last_run = { "next_run_earliest_time": "2018-10-24T14:13:20.000+00:00", "late_indexed_pagination": True, "next_run_found_incidents_ids": { "1234": {"occurred_time": "2018-10-24T14:10:20.000+00:00"}, "5678": {"occurred_time": "2018-10-24T14:10:20.000+00:00"}, }, } mock_params = {"fetchQuery": "something"} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=[self.finding1]) service = self.Service() oneshot_mocker = mocker.patch.object(service.jobs, "oneshot", side_effect=service.jobs.oneshot) mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper) assert oneshot_mocker.call_args_list[0][0][0] == 'something | where not event_id in ("1234","5678")' assert oneshot_mocker.call_args_list[0][1]["offset"] == 0 # If (num_of_dropped == FETCH_LIMIT and '`notable`' in fetch_query), then late_indexed_pagination should be set to True def test_first_condition_for_late_indexed_pagination(self, mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch): """ Given - Incident IDs that were fetched in the last fetch round - Mocked incidents api response, that have IDs as the last fetched IDs (which means that num_of_dropped == FETCH_LIMIT) - `notable` is in the fetch query When - Fetching findings Then - Make sure that the key "late_indexed_pagination" in last run object is set to True """ from SplunkPyV2 import UserMappingObject # MonkeyPatch can be used to patch global variables monkeypatch.setattr(splunk, "FETCH_LIMIT", 2) mocker.patch.object(demisto, "setLastRun") mock_last_run = { "next_run_earliest_time": "2018-10-24T14:13:20.000+00:00", "late_indexed_pagination": True, "next_run_found_incidents_ids": { "id_1": {"occurred_time": "2018-10-24T14:10:20.000+00:00"}, "id_2": {"occurred_time": "2018-10-24T14:1:20.000+00:00"}, }, } mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mock_params = {"fetchQuery": "`notable` is cool", "max_fetch": 2} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=[self.finding1, self.finding2]) set_last_run_mocker = mocker.patch("demistomock.setLastRun") service = self.Service() mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper) assert set_last_run_mocker.call_args_list[0][0][0]["late_indexed_pagination"] is True # If (len(incidents) == FETCH_LIMIT and late_indexed_pagination), then late_indexed_pagination should be set to True def test_second_condition_for_late_indexed_pagination(self, mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch): """ Given - Incident IDs that were fetched in the last fetch round - Mocked incidents api response, where only new incidents are fetched (which means that len(incidents) == FETCH_LIMIT) - The key "late_indexed_pagination" in the last run object is set to True When - Fetching findings Then - Make sure that the key "late_indexed_pagination" in last run object is set to True """ from SplunkPyV2 import UserMappingObject # MonkeyPatch can be used to patch global variables monkeypatch.setattr(splunk, "FETCH_LIMIT", 2) mocker.patch.object(demisto, "setLastRun") mock_last_run = { "next_run_earliest_time": "2018-10-24T14:13:20.000+00:00", "late_indexed_pagination": True, "next_run_found_incidents_ids": { "1234": {"occurred_time": "2018-10-24T14:10:20.000+00:00"}, "5678": {"occurred_time": "2018-10-24T14:1:20.000+00:00"}, }, } mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mock_params = {"fetchQuery": "`notable` is cool", "max_fetch": 2} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=[self.finding1, self.finding2]) set_last_run_mocker = mocker.patch("demistomock.setLastRun") service = self.Service() mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper) assert set_last_run_mocker.call_args_list[0][0][0]["late_indexed_pagination"] is True def test_fetch_incidents(mocker): """ Given - mocked incidents api response - a mapper which should not map the user owner into the incident response When - executing the fetch incidents flow Then - make sure the incident response is valid. - make sure that the owner is not part of the incident response """ from SplunkPyV2 import UserMappingObject mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mocker.patch.object(demisto, "incidents") mocker.patch.object(demisto, "setLastRun") mock_last_run = {"time": "2018-10-24T14:13:20"} mock_params = {"fetchQuery": "something"} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) service = mocker.patch("splunklib.client.connect", return_value=None) mocker.patch("splunklib.results.JSONResultsReader", return_value=deepcopy(SAMPLE_RESPONSE)) mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper) incidents = demisto.incidents.call_args[0][0] assert demisto.incidents.call_count == 1 assert len(incidents) == 2 assert incidents[0]["name"] == "Endpoint - Recurring Malware Infection - Rule : Endpoint - Recurring Malware Infection - Rule" assert not incidents[0].get("owner") SPLUNK_RESULTS = [ { "rawJSON": '{"source": "This is the alert type", "field_name1": "field_val1", "field_name2": "field_val2"}', "details": "Endpoint - High Or Critical Priority Host With Malware - Rule", "labels": [{"type": "security_domain", "value": "Endpoint - High Or Critical Priority Host With Malware - Rule"}], } ] EXPECTED_OUTPUT = { "This is the alert type": {"source": "This is the alert type", "field_name1": "field_val1", "field_name2": "field_val2"} } def test_create_mapping_dict(): mapping_dict = splunk.create_mapping_dict(SPLUNK_RESULTS, type_field="source") assert mapping_dict == EXPECTED_OUTPUT def test_fetch_findings(mocker): """ Given - mocked incidents api response - a mapper which should not map the user owner into the incident response When - executing the fetch findings flow Then - make sure the incident response is valid. - make sure that the owner is not part of the incident response """ mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mocker.patch.object(splunk.client.Job, "is_done", return_value=True) mocker.patch.object(splunk.client.Job, "results", return_value=None) mocker.patch.object( splunk, "ENABLED_ENRICHMENTS", [splunk.ASSET_ENRICHMENT, splunk.DRILLDOWN_ENRICHMENT, splunk.IDENTITY_ENRICHMENT] ) mocker.patch.object(demisto, "incidents") mocker.patch.object(demisto, "setLastRun") mock_last_run = {"time": "2018-10-24T14:13:20"} mock_params = {"fetchQuery": "something"} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) service = Service("DONE") mocker.patch("splunklib.results.JSONResultsReader", return_value=deepcopy(SAMPLE_RESPONSE)) mapper = splunk.UserMappingObject(service, False) splunk.fetch_incidents(service, mapper=mapper) cache_object = splunk.Cache.load_from_integration_context(get_integration_context()) assert cache_object.submitted_findings finding = cache_object.submitted_findings[0] incident_from_cache = finding.to_incident(mapper) incidents = demisto.incidents.call_args[0][0] assert demisto.incidents.call_count == 1 assert len(incidents) == 0 assert ( incident_from_cache["name"] == "Endpoint - Recurring Malware Infection - Rule : Endpoint - " "Recurring Malware Infection - Rule" ) assert not incident_from_cache.get("owner") # now call second time to make sure that the incident fetched splunk.fetch_incidents(service, mapper=mapper) incidents = demisto.incidents.call_args[0][0] assert len(incidents) == 2 assert incidents[0]["name"] == "Endpoint - Recurring Malware Infection - Rule : Endpoint - Recurring Malware Infection - Rule" assert not incidents[0].get("owner") def test_fetch_findings_with_creation_time1(mocker: MockerFixture): """ Given: A configuration using "creation time" as the finding time source in demisto parameters. When: The fetch_findings function is called. Then: The function should query Splunk using the earliest_time and latest_time fields in the search kwargs. """ mocker.patch.object( demisto, "params", return_value={"finding_time_source": "creation time", "fetchQuery": "something", "occurrence_look_behind": "0"}, ) mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mocker.patch.object(results, "JSONResultsReader", return_value=[]) # Mock the service object mock_service = mocker.MagicMock() mock_search = mocker.MagicMock() mock_service.jobs.oneshot.return_value = mock_search # Mock the search results mock_search.results = mocker.MagicMock(return_value=[]) # Mock the mapper object mock_mapper = mocker.MagicMock() # Create a mock for the Cache mock_cache = mocker.MagicMock() # Call the function splunk.fetch_findings( service=mock_service, mapper=mock_mapper, cache_object=mock_cache, enrich_findings=False, ) # Verify that the service.jobs.oneshot was called with "creation time" in the kwargs call_args = mock_service.jobs.oneshot.call_args[1] # The query should include "creation time" in the search criteria assert "earliest_time" in call_args assert "latest_time" in call_args assert "index_earliest" not in call_args assert "index_latest" not in call_args def test_fetch_findings_with_index_time1(mocker: MockerFixture): """ Given: A configuration using "index time" as the finding time source in demisto parameters. When: The fetch_findings function is called. Then: The function should query Splunk using the index_earliest and index_latest fields in the search kwargs. """ mocker.patch.object( demisto, "params", return_value={"finding_time_source": "index time", "fetchQuery": "something", "occurrence_look_behind": "0"}, ) mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mocker.patch.object(splunk.client.Job, "is_done", return_value=True) mocker.patch.object(results, "JSONResultsReader", return_value=[]) # Mock the service object mock_service = mocker.MagicMock() mock_search = mocker.MagicMock() mock_service.jobs.oneshot.return_value = mock_search # Mock the search results mock_search.results = mocker.MagicMock(return_value=[]) # Mock the mapper object mock_mapper = mocker.MagicMock() # Create a mock for the Cache mock_cache = mocker.MagicMock() # Call the function splunk.fetch_findings( service=mock_service, mapper=mock_mapper, cache_object=mock_cache, enrich_findings=False, ) # Verify that the service.jobs.oneshot was called with "creation time" in the kwargs call_args = mock_service.jobs.oneshot.call_args[1] # The query should include "creation time" in the search criteria assert "index_earliest" in call_args assert "index_latest" in call_args assert "earliest_time" not in call_args assert "latest_time" not in call_args """ ========== Enriching Fetch Mechanism Tests ========== """ @pytest.mark.parametrize( "integration_context, output", [({splunk.INCIDENTS: ["incident"]}, ["incident"]), ({splunk.INCIDENTS: []}, []), ({}, [])] ) def test_fetch_incidents_for_mapping(integration_context, output, mocker): """ Scenario: When a user configures a mapper using Fetch from Instance when the enrichment mechanism is working, we save the ready incidents in the integration context. Given: - List of ready incidents - An empty list of incidents - An empty integration context object When: - fetch_incidents_for_mapping is called Then: - Return the expected result """ mocker.patch.object(demisto, "info") mocker.patch.object(demisto, "incidents") splunk.fetch_incidents_for_mapping(integration_context) assert demisto.incidents.call_count == 1 assert demisto.incidents.call_args[0][0] == output def test_reset_enriching_fetch_mechanism(mocker): """ Scenario: When a user is willing to reset the enriching fetch mechanism and start over. Given: - An integration context object with not empty Cache and incidents When: - reset_enriching_fetch_mechanism is called Then: - Check that the integration context does not contain this fields """ set_mocker = mocker.patch("SplunkPyV2.set_integration_context") splunk.reset_enriching_fetch_mechanism() assert set_mocker.call_args[0][0] == {} def test_given_enrichment_enabled_when_fetch_then_dedup_ids_persisted_in_final_setLastRun(mocker): """ Regression test for the dedup-IDs-lost-when-enrichment-enabled bug. Given: - Enrichments are enabled (so the FindingsFetchHandler routes through ``run_enrichment_mechanism`` instead of the plain ``fetch_findings`` path). - ``run_enrichment_mechanism`` is stubbed to return a non-empty ``last_run_delta`` containing the dedup keys (``next_run_found_incidents_ids``, ``next_run_earliest_time``). When: - The dispatcher ``fetch_incidents`` runs one cycle. Then: - The final ``demisto.setLastRun`` (the dispatcher's single emit at the end of the cycle) must carry the dedup IDs and time cursor produced by the enrichment mechanism. Previously, the dispatcher snapshotted ``demisto.getLastRun()`` BEFORE invoking the handler and then re-wrote that stale snapshot at the end of the cycle, wiping out any inner ``setLastRun`` performed by ``run_enrichment_mechanism``. The fix lets the enrichment path return its delta through ``FetchResult.last_run_delta`` so the dispatcher merges and persists it in the final emit. """ # GIVEN — enrichment enabled and stable last-run/params plumbing. mocker.patch.object(splunk, "ENABLED_ENRICHMENTS", new=[splunk.DRILLDOWN_ENRICHMENT]) mocker.patch.object(demisto, "getLastRun", return_value={"next_run_earliest_time": "2018-10-24T14:13:20.000+00:00"}) mocker.patch.object(demisto, "params", return_value={"fetch_event_types": "Finding"}) mocker.patch.object(demisto, "incidents") set_last_run_mock = mocker.patch.object(demisto, "setLastRun") mocker.patch.object(splunk, "get_integration_context", return_value={}) # The enrichment helper returns (incidents, last_run_delta). We surface a # delta with the dedup keys we expect to see at the end of the cycle. expected_dedup_ids = {"event-id-1": {"occurred_time": "2018-10-24T14:23:20.000+00:00"}} expected_delta = { "next_run_earliest_time": "2018-10-24T14:23:20.000+00:00", "next_run_latest_time": None, "offset": 0, "next_run_found_incidents_ids": expected_dedup_ids, splunk.DUMMY: splunk.DUMMY, } mocker.patch.object(splunk, "run_enrichment_mechanism", return_value=([], expected_delta)) # WHEN — dispatcher runs one cycle. service = mocker.MagicMock() mapper = splunk.UserMappingObject(service, False) splunk.fetch_incidents(service, mapper=mapper) # THEN — the final setLastRun must carry the dedup keys returned by the # enrichment mechanism (i.e. the dispatcher merged them in, not overwrote them). assert set_last_run_mock.call_count == 1 final_last_run = set_last_run_mock.call_args[0][0] assert final_last_run["next_run_found_incidents_ids"] == expected_dedup_ids assert final_last_run["next_run_earliest_time"] == "2018-10-24T14:23:20.000+00:00" assert final_last_run["offset"] == 0 assert final_last_run.get(splunk.DUMMY) == splunk.DUMMY @pytest.mark.parametrize( "drilldown_creation_time, asset_creation_time, enrichment_timeout, output", [ # Case 1: Both enrichments are recent (within timeout) ("2025-12-09T20:00:00.000000+00:00", "2025-12-09T20:00:00.000000+00:00", 5, False), # Case 2: Drilldown enrichment is older than timeout, asset enrichment is recent ("2025-12-09T19:54:00.000000+00:00", "2025-12-09T20:00:00.000000+00:00", 5, True), ], ) def test_is_enrichment_exceeding_timeout(mocker, drilldown_creation_time, asset_creation_time, enrichment_timeout, output): """ Scenario: When one of the finding's enrichments is exceeding the timeout, we want to create an incident with all the data gathered so far. Given: - Two enrichments that none of them exceeds the timeout. - An enrichment exceeding the timeout and one that does not exceeds the timeout. When: - is_enrichment_process_exceeding_timeout is called Then: - Return the expected result """ mocked_dt = mocker.patch("SplunkPyV2.datetime") mocked_dt.now.return_value = datetime.strptime("2025-12-09T20:01:00.000000+00:00", splunk.ISO_FORMAT_TZ_AWARE) mocked_dt.strptime.side_effect = datetime.strptime mocker.patch.object(splunk, "ENABLED_ENRICHMENTS", return_value=[splunk.DRILLDOWN_ENRICHMENT, splunk.ASSET_ENRICHMENT]) finding = splunk.Finding({splunk.EVENT_ID: "id"}) finding.enrichments.append(splunk.Enrichment(splunk.DRILLDOWN_ENRICHMENT, creation_time=drilldown_creation_time)) finding.enrichments.append(splunk.Enrichment(splunk.ASSET_ENRICHMENT, creation_time=asset_creation_time)) assert finding.is_enrichment_process_exceeding_timeout(enrichment_timeout) is output INCIDENT_1 = {"name": "incident1", "rawJSON": json.dumps({})} INCIDENT_2 = {"name": "incident2", "rawJSON": json.dumps({})} @pytest.mark.parametrize("incidents, output", [([], []), ([INCIDENT_1, INCIDENT_2], [INCIDENT_1, INCIDENT_2])]) def test_store_incidents_for_mapping(incidents, output): """ Scenario: Store ready incidents in integration context, to be retrieved by a user configuring a mapper and selecting "Fetch from instance" when the enrichment mechanism is working. Given: - An empty list of incidents - A list of two incidents When: - store_incidents_for_mapping is called Then: - Return the expected result """ splunk.set_integration_context({}) splunk.store_incidents_for_mapping(incidents) assert splunk.get_integration_context().get(splunk.INCIDENTS, []) == output @pytest.mark.parametrize( "finding_data, raw, earliest, latest", [ ({}, {}, "", ""), ( {"drilldown_earliest": f"${splunk.INFO_MIN_TIME}$", "drilldown_latest": f"${splunk.INFO_MAX_TIME}$"}, {splunk.INFO_MIN_TIME: "1", splunk.INFO_MAX_TIME: "2"}, "1", "2", ), ( { "drilldown_earliest": "1", "drilldown_latest": "2", }, {}, "1", "2", ), ], ) def test_get_drilldown_timeframe(finding_data, raw, earliest, latest, mocker): """ Scenario: Trying to get the drilldown's timeframe from the finding's data Given: - An empty finding's data - An finding's data that the info of the timeframe is in the raw field - An finding's data that the info is in the data dict When: - get_drilldown_timeframe is called Then: - Return the expected result """ mocker.patch.object(demisto, "info") earliest_offset, latest_offset = splunk.get_drilldown_timeframe(finding_data, raw) assert earliest_offset == earliest assert latest_offset == latest @pytest.mark.parametrize( "raw_field, finding_data, expected_field, expected_value", [ ("field|s", {"field": "1"}, "field", "1"), ("field", {"field": "1"}, "field", "1"), ("field|s", {"_raw": "field=1, value=2"}, "field", "1"), ("x", {"y": "2"}, "", ""), # A raw field that is a substring of another field must not collide ("src_ip", {"src": "host1", "src_ip": "1.2.3.4"}, "src_ip", "1.2.3.4"), ("src_ip|s", {"src": "host1", "src_ip": "1.2.3.4"}, "src_ip", "1.2.3.4"), ("src_ip", {"_raw": "src=host1, src_ip=1.2.3.4"}, "src_ip", "1.2.3.4"), ], ) def test_get_finding_field_and_value(raw_field, finding_data, expected_field, expected_value, mocker): """ Scenario: When building the drilldown search query, we search for the field in the raw search query and search for its real name in the finding's data or in the finding's raw data. We also ignore Splunk advanced syntax such as "|s, |h, ..." Given: - A raw field that has the same name in the finding's data - A raw field that has "|s" as a suffix in the raw search query and its value is in the finding's data - A raw field that has "|s" as a suffix in the raw search query and its value is in the finding's raw data - A raw field that is not is the finding's data or in the finding's raw data When: - get_finding_field_and_value is called Then: - Return the expected result """ mocker.patch.object(demisto, "error") field, value = splunk.get_finding_field_and_value(raw_field, finding_data) assert field == expected_field assert value == expected_value @pytest.mark.parametrize( "finding_data, search, raw, is_query_name, expected_search", [ ({"a": "1", "_raw": "c=3"}, "search a=$a|s$ c=$c$ suffix", {"c": "3"}, False, 'search a="1" c="3" suffix'), ({"a": ["1", "2"], "b": "3"}, "search a=$a|s$ b=$b|s$ suffix", {}, False, 'search (a="1" OR a="2") b="3" suffix'), ({"a": "1", "_raw": "b=3", "event_id": "123"}, "search a=$a|s$ c=$c$ suffix", {"b": "3"}, False, ""), ( {"signature": "Backdoor.test"}, "View related '$signature$' events for $dest$", {"dest": "ACME-test-005"}, True, "View related 'Backdoor.test' events for ACME-test-005", ), ( {}, 'View all wineventlogs involving user="$user$"', {"user": "test"}, True, 'View all wineventlogs involving user="test"', ), ({}, "Test query name", {}, True, "Test query name"), ( {"user": "test\\crusher"}, 'index="test" | where user = $user|s$', {}, False, 'index="test" | where user="test\\\\crusher"', ), ( {"user": "test\\crusher"}, 'index="test" | where user = "$user|s$"', {}, False, 'index="test" | where user="test\\\\crusher"', ), ( {"countryNameA": '"test\\country"', "countryNameB": '""'}, 'search countryA="$countryNameA|s$" countryB=$countryNameB|s$', {}, False, 'search countryA="test\\country" countryB=""', ), ({"test": "test_user"}, "search countryA=\\$this is a test\\$", {}, False, "search countryA=\\$this is a test\\$"), # A field whose name is a substring of the queried field must resolve to the correct field ( {"src": "host1", "src_ip": "1.2.3.4"}, "search src_ip=$src_ip$", {}, False, 'search src_ip="1.2.3.4"', ), ], ids=[ "search query fields in findings data and raw data", "search query fields in finding data more than one value", "search query fields don't exist in finding data and raw data", "query name fields in findings data and raw data", "query name fields in raw data", "query name without fields to replace", "search query with a user field that contains a backslash", "search query with a user field that is surrounded by quotation marks and contains a backslash", "search query fields in finding data more than one value, with one empty value", "search query with $ as part of the search - no need to replace", "search query with a field name that is a substring of another field ", ], ) def test_build_drilldown_search(finding_data, search, raw, is_query_name, expected_search, mocker): """ Scenario: When building the drilldown search query, we replace every field in between "$" sign with its corresponding query part (key & value). Given: - A raw search query with fields both in the finding's data and in the finding's raw data - A raw search query with fields in the finding's data that has more than one value - A raw search query with fields that does not exist in the finding's data or in the finding's raw data - A raw query name with fields both in the finding's data and in the finding's raw data - A raw query name with fields in the finding's raw data - A raw query name without any fields to replace. - A raw query search with a user field that contains a backslash - A raw query search with a user field that is surrounded by quotation marks and contains a backslash When: - build_drilldown_search is called Then: - Return the expected result """ mocker.patch.object(demisto, "error") mocker.patch.object(demisto, "params", return_value={}) parsed_query = splunk.build_drilldown_search(finding_data, search, raw, is_query_name) assert parsed_query == expected_search @pytest.mark.parametrize( "finding_data, prefix, fields, query_part", [ ({"user": ["u1", "u2"]}, "identity", ["user"], '(identity="u1" OR identity="u2")'), ({"_raw": "1233, user=u1"}, "user", ["user"], 'user="u1"'), ( {"user": ["u1", "u2"], "_raw": "1321, src_user=u3"}, "user", ["user", "src_user"], '(user="u1" OR user="u2" OR user="u3")', ), ({}, "prefix", ["field"], ""), ], ) def test_get_fields_query_part(finding_data, prefix, fields, query_part): """ Scenario: When building an enrichment search query, we search for values in the finding's data / finding's raw data and fill them in the raw search query to create a searchable query. Given: - One field with multiple values, values in the data - One field, value is in the raw data - Two fields with multiple values, values in both the data and the raw data - An empty finding data, field does not exists When: - get_fields_query_part is called Then: - Return the expected result """ assert splunk.get_fields_query_part(finding_data, prefix, fields) == query_part @pytest.mark.parametrize( "enrichments, expected_data", [ ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name1", query_search="query_search1", data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="2", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name2", query_search="query_search2", data=[{"result1": "c"}, {"result2": "d"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="3", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name3", query_search="query_search3", data=[{"result1": "e"}, {"result2": "f"}], ), ], [ { "query_name": "query_name1", "query_search": "query_search1", "query_results": [{"result1": "a"}, {"result2": "b"}], "enrichment_status": splunk.Enrichment.SUCCESSFUL, }, { "query_name": "query_name2", "query_search": "query_search2", "query_results": [{"result1": "c"}, {"result2": "d"}], "enrichment_status": splunk.Enrichment.SUCCESSFUL, }, { "query_name": "query_name3", "query_search": "query_search3", "query_results": [{"result1": "e"}, {"result2": "f"}], "enrichment_status": splunk.Enrichment.SUCCESSFUL, }, ], ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name1", query_search="query_search1", data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="2", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name2", query_search="query_search2", data=[{"result1": "c"}, {"result2": "d"}], ), ], [ { "query_name": "query_name1", "query_search": "query_search1", "query_results": [{"result1": "a"}, {"result2": "b"}], "enrichment_status": splunk.Enrichment.SUCCESSFUL, }, { "query_name": "query_name2", "query_search": "query_search2", "query_results": [{"result1": "c"}, {"result2": "d"}], "enrichment_status": splunk.Enrichment.SUCCESSFUL, }, ], ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name1", query_search="query_search1", data=[{"result1": "a"}, {"result2": "b"}], ) ], [ { "query_name": "query_name1", "query_search": "query_search1", "query_results": [{"result1": "a"}, {"result2": "b"}], "enrichment_status": splunk.Enrichment.SUCCESSFUL, } ], ), ([], None), ], ids=[ "A Finding with 3 drilldown enrichments, 1 asset enrichment and 1 identity enrichment", "A Finding with 2 drilldown enrichment, 1 asset enrichment and 1 identity enrichment", "A Finding with 1 drilldown enrichment, 1 asset enrichment and 1 identity enrichment", "A Finding without drilldown enrichments, 1 asset enrichments and 1 identity enrichment", ], ) def test_to_incident_finding_enrichments_data(enrichments, expected_data): """ Tests the logic of the Finding.to_incident() function, regarding the results data of multiple drilldown enrichments. Given: 1. A Finding with 3 drilldown enrichments, 1 asset enrichment and 1 identity enrichment. 2. A Finding with 2 drilldown enrichment, 1 asset enrichment and 1 identity enrichment. 3. A Finding with 1 drilldown enrichment, 1 asset enrichment and 1 identity enrichment. 4. A Finding without drilldown enrichments, 1 asset enrichments and 1 identity enrichment. When: - Finding.to_incident() function is called Then: - Verify that the data of the finding includes the expected enrichements result as follow: 1. A dictionary with the results of the 3 drilldown searches by query names. 2. A dictionary with the results of the 2 drilldown searches by query names. 3. A list of the drilldown searches results (backwards competability). 4. No 'Drilldown' key in the findings data. """ finding = splunk.Finding({}, finding_id="id", enrichments=enrichments) enrichments_to_add = [ splunk.Enrichment( splunk.ASSET_ENRICHMENT, enrichment_id="111", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.IDENTITY_ENRICHMENT, enrichment_id="222", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), ] finding.enrichments.extend(enrichments_to_add) service = Service("DONE") mapper = splunk.UserMappingObject(service, False) finding.to_incident(mapper) assert finding.data.get(splunk.ASSET_ENRICHMENT) == [{"result1": "a"}, {"result2": "b"}] assert finding.data.get(splunk.IDENTITY_ENRICHMENT) == [{"result1": "a"}, {"result2": "b"}] assert finding.data.get(splunk.DRILLDOWN_ENRICHMENT) == expected_data @pytest.mark.parametrize( "enrichments, enrichment_type, expected_stauts_result", [ ( [ splunk.Enrichment( splunk.ASSET_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.ASSET_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.ASSET_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.ASSET_ENRICHMENT, False, ), ( [ splunk.Enrichment( splunk.IDENTITY_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.IDENTITY_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.IDENTITY_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.IDENTITY_ENRICHMENT, False, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name1", query_search="query_search1", data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.DRILLDOWN_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, query_name="query_name1", query_search="query_search1", data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.DRILLDOWN_ENRICHMENT, False, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), ], splunk.DRILLDOWN_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), ], splunk.DRILLDOWN_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), ], splunk.DRILLDOWN_ENRICHMENT, False, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), ], splunk.DRILLDOWN_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), ], splunk.DRILLDOWN_ENRICHMENT, True, ), ], ids=[ "A Finding with 1 successful Asset enrichment", "A Finding with 1 failed Asset enrichment", "A Finding with 1 successful Identity enrichment", "A Finding with 1 failed Identity enrichment", "A Finding with 1 successful Drilldown enrichment", "A Finding with 1 failed Drilldown enrichment", "A Finding with 1 successful Drilldown enrichment and 1 failed drilldown enrichment (the first is successful)", "A Finding with 1 successful Drilldown enrichment and 1 failed drilldown enrichment (the second is successful)", "A Finding with 2 Drilldown enrichments [failed, failed]", "A Finding with 2 Drilldown enrichments [successful, successful]", "A Finding with 3 Drilldown enrichments [failed, successful, failed]", ], ) def test_to_incident_finding_enrichments_status(enrichments, enrichment_type, expected_stauts_result): """ Tests the logic of the Finding.to_incident() function, regarding the statuses of enrichments. Given: 1. A Finding with 1 successful Asset enrichment. 2. A Finding with 1 failed Asset enrichment. 3. A Finding with 1 successful Identity enrichment. 4. A Finding with 1 failed Identity enrichment. 5. A Finding with 1 successful Drilldown enrichment. 6. A Finding with 1 failed Drilldown enrichment. 7. A Finding with 1 successful Drilldown enrichment and 1 failed drilldown enrichment (the first is successful). 8. A Finding with 1 successful Drilldown enrichment and 1 failed drilldown enrichment (the second is successful). 9. A Finding with 2 Drilldown enrichments [failed, failed]. 10. A Finding with 2 Drilldown enrichments [successful, successful]. 11. A Finding with 3 Drilldown enrichments [failed, successful, failed]. When: - Finding.to_incident() function is called Then: - Verify that the status of the finding enrichments is as follow: 1. Asset Enrichment status is: successful_asset_enrichment = True. 2. Asset Enrichment status is: successful_asset_enrichment = False. 3. Identity Enrichment status is: successful_identity_enrichment = True. 4. Identity Enrichment status is: successful_identity_enrichment = False. # In Drilldown enrichment - if at least one drilldown enrichment is successful the status is Success. 5. Drilldown Enrichment status is: successful_drilldown_enrichment = True. 6. Drilldown Enrichment status is: successful_drilldown_enrichment = False. 7. Drilldown Enrichment status is: successful_drilldown_enrichment = True. 8. Drilldown Enrichment status is: successful_drilldown_enrichment = True. 9. Drilldown Enrichment status is: successful_drilldown_enrichment = False. 10. Drilldown Enrichment status is: successful_drilldown_enrichment = True. 11. Drilldown Enrichment status is: successful_drilldown_enrichment = True. """ finding = splunk.Finding({}, finding_id="id", enrichments=enrichments) service = Service("DONE") mapper = splunk.UserMappingObject(service, False) finding.to_incident(mapper) assert finding.data[splunk.ENRICHMENT_TYPE_TO_ENRICHMENT_STATUS[enrichment_type]] == expected_stauts_result @pytest.mark.parametrize( "spl_search, expected", [ # Single backslashes inside a field="value" filter must be doubled ( 'eventcode IN (1, 2) field_a="\\foo\\bar\\baz" | head 1', 'eventcode IN (1, 2) field_a="\\\\foo\\\\bar\\\\baz" | head 1', ), # Already-doubled values are left unchanged (idempotent) ('field_a="\\\\foo\\\\bar"', 'field_a="\\\\foo\\\\bar"'), # Values without backslashes are untouched ('field_a="10.0.0.1" field_b="abc"', 'field_a="10.0.0.1" field_b="abc"'), # rex / free-text quoted strings (not preceded by '=') must NOT be modified ( 'index=x | rex field=field_a "value: (?.*)"', 'index=x | rex field=field_a "value: (?.*)"', ), # Regex literals inside SPL function calls (quote follows '(' or ',', NOT a field token) # must NOT be re-escaped. ( '| eval field_a=replace(field_a,"(\\\\)","\\\\\\\\")', '| eval field_a=replace(field_a,"(\\\\)","\\\\\\\\")', ), # Multiple genuine field="value" filters in one query are all doubled ( 'field_a="\\foo\\bar" field_b="\\\\baz\\\\qux"', 'field_a="\\\\foo\\\\bar" field_b="\\\\baz\\\\qux"', ), # A dotted field name is still treated as a field filter ( 'parent.child="\\foo\\bar"', 'parent.child="\\\\foo\\\\bar"', ), ], ids=[ "single backslashes are doubled", "already-doubled backslashes are unchanged", "no backslashes are untouched", "rex regex quoted string is not modified", "eval/replace regex literal is not over-escaped", "multiple field filters are all doubled", "dotted field name is treated as a field filter", ], ) def test_escape_backslashes_in_field_filters(spl_search, expected): """ Scenario: A drilldown search arrives as JSON; after json.loads, backslashes inside field filter values are collapsed to single backslashes, which Splunk SPL cannot match. We re-escape them. Given: - An SPL search with a field="value" filter value containing single backslashes. - An SPL search whose field filter values are already correctly escaped. - An SPL search without backslashes. - An SPL search with a rex/free-text quoted string containing a backslash. - An SPL search with a regex literal inside a function call (eval/replace). - An SPL search with multiple field filters. - An SPL search with a dotted field name. When: - escape_backslashes_in_field_filters is called. Then: - Backslashes inside genuine field="value" filters are doubled, the operation is idempotent, and rex/free-text quoted strings and regex literals inside function calls are left untouched. """ assert splunk.escape_backslashes_in_field_filters(spl_search) == expected def test_parse_drilldown_searches_preserves_backslashes(): """ Given: - A 'drilldown_searches' JSON payload where a field="value" filter value contains backslashes. When: - Running splunk.parse_drilldown_searches. Then: - The parsed 'search' keeps the backslashes escaped (doubled) for Splunk SPL. """ searches = [ '[{"name":"Show events","search":"(index=idx_a OR index=idx_b) ' 'eventcode IN (1, 2) field_a=\\"\\\\foo\\\\bar\\\\baz\\" ' '| head 1","earliest_offset":"1","latest_offset":"2","disabled":false}]' ] parsed = splunk.parse_drilldown_searches(searches) assert parsed[0]["search"] == ("(index=idx_a OR index=idx_b) eventcode IN (1, 2) " 'field_a="\\\\foo\\\\bar\\\\baz" | head 1') def test_parse_drilldown_searches(): """ Given: - A list of valid Json strings with splunk drilldown searches data. When: - Running the splunk.parse_drilldown_searches function Then: - Verify that the search data was parsed into a python dictionary as expected. """ searches = [ '{"name":"View related \'$signature$\' events for $dest$","search":"| from datamodel:\\"Malware\\".' '\\"Malware_Attacks\\" | search dest=$dest|s$ signature=$signature|s$","earliest":17145' '63300,"latest":1715168700}', '{"name":"View related \'$category$\' events for $signature$","search":"| from datamodel:\\"Malw' 'are\\".\\"Malware_Attacks\\" \\n| fields category, dest, signature | search dest=$dest|s$ signature=' '$signature|s$","earliest":1714563300,"latest":1715168700}', ] parsed_searches = splunk.parse_drilldown_searches(searches) for search in parsed_searches: assert isinstance(search, dict) assert parsed_searches == [ { "name": "View related '$signature$' events for $dest$", "search": '| from datamodel:"Malware"."Malware_Attacks" | search dest=$dest|s$ signature=$signature|s$', "earliest": 1714563300, "latest": 1715168700, }, { "name": "View related '$category$' events for $signature$", "search": '| from datamodel:"Malware"."Malware_Attacks" \n| fields category, dest, signature | search dest=$dest|s$ ' "signature=$signature|s$", "earliest": 1714563300, "latest": 1715168700, }, ] @pytest.mark.parametrize( "finding_data, expected_call_count", [ ({"event_id": "test_id", "drilldown_search": "test_search", "drilldown_searches": ["test_search1", "test_search2"]}, 0), ({"event_id": "test_id", "drilldown_search": "", "drilldown_searches": ["test_search1", "test_search2"]}, 1), ({"event_id": "test_id", "drilldown_searches": ["test_search1", "test_search2"]}, 1), ], ids=[ "A finding data with both 'drilldown_search' and 'drilldown_searches' keys with values", "A finding data with both 'drilldown_search' and 'drilldown_searches' keys but 'drilldown_search' has no value", "A finding data with 'drilldown_searches' key only", ], ) def test_drilldown_enrichment_main_condition(mocker, finding_data, expected_call_count): """ Tests the logic of the first (main) condition in the drilldown_enrichment() function. We want to make sure that in a case that the finding data include both 'drilldown_search' and 'drilldown_searches' keys (happens when there is only one drilldown search to enrich) the 'drilldown_search' value will be taken to maintain backwards cometability. In any other case the value of the 'drilldown_searches' key will be used. Given: 1. A finding data that includes both 'drilldown_search' and 'drilldown_searches' keys with values. 2. A finding data that includes both 'drilldown_search' and 'drilldown_searches' keys but 'drilldown_search' has no value. 3. A finding data that includes 'drilldown_searches' key only. When: - Running the splunk.drilldown_enrichment function Then: - Verify that: 1. The value of the 'drilldown_search' key is taken (to maintain backwards competability), and therefore we don't call the parse_drilldown_searches function. 2. The value of the 'drilldown_searches' key is taken, and therefore we call the parse_drilldown_searches function. 3. The value of the 'drilldown_searches' key is taken, and therefore we call the parse_drilldown_searches function. """ mock_parse_drilldown_searches = mocker.patch("SplunkPyV2.parse_drilldown_searches", return_value=[]) service = Service("DONE") splunk.drilldown_enrichment(service, finding_data, 5) assert mock_parse_drilldown_searches.call_count == expected_call_count @pytest.mark.parametrize( "finding_data, expected_call_count", [ ({"event_id": "test_id", "drilldown_search": "test_search", "drilldown_searches": [{}], "_raw": "{'test':1}"}, 1), ( { "event_id": "test_id", "drilldown_searches": [ '{"name":"View related \'$signature$\' events for $dest$","search":"| from datamodel:\\"Malware\\".\\"Malwa' 're_Attacks\\" | search dest=$dest|s$ signature=$signature|s$","earliest":1714563300,"latest":1715168700}', '{"name":"View related \'$category$\' events for $signature$","search":"| from datamodel:\\"Malware\\".\\"M' 'alware_Attacks\\" \\n| fields category, dest, signature | search dest=$dest|s$ signature=$signature|s$",' '"earliest":1714563300,"latest":1715168700}', ], }, 0, ), ], ids=["A finding data with one drilldown search", "A finding data with multiple drilldown searches"], ) def test_drilldown_enrichment_get_timeframe(mocker, finding_data, expected_call_count): """ Tests that in a case of one drildown search we extract the search timeframe from the finding data by calling the get_drilldown_timeframe() function, and in a case of multiple drilldown searches, we get the timeframe from the drilldown search data dictionary without calling the get_drilldown_timeframe() function. Given: 1. A finding data with one drilldown search. 2. A finding data with multiple drilldown searches. When: - Running the splunk.get_drilldown_timeframe function. Then: - Verify that: 1. The timeframe is determined according to fields in the finding data and raw data by using the get_drilldown_timeframe function. 2. The timeframe is determined according to fields of each drilldown search data dict. """ mock_get_drilldown_timeframe = mocker.patch("SplunkPyV2.get_drilldown_timeframe", return_value=("", "")) mocker.patch("SplunkPyV2.build_drilldown_search", return_value="") service = Service("DONE") splunk.drilldown_enrichment(service, finding_data, 5) assert mock_get_drilldown_timeframe.call_count == expected_call_count def test_drilldown_enrichment_query_earliest(mocker: MockerFixture): """ Tests the drilldown enrichment process when query contains earliest filter Given: A drilldown data without drilldown_earliest and drilldown_latest values, the drilldown search query contains earliest filter. When: Performing drilldown enrichment to generate search jobs and queries Then: The generated query match the expected enriched query and contains then earliest value """ from splunklib import client service = Service("DONE") mock_params = {"fetchQuery": "`notable` is cool | fillnull value=NULL"} mocker.patch("demistomock.params", return_value=mock_params) notable_data = { "event_id": "test_id", "drilldown_name": "View all login attempts by system $src$", "drilldown_search": '| from datamodel:"Authentication"."Authentication" | search src=$src|s$ | earliest=1d', "drilldown_searches": "[]", "_raw": "src='test_src'", } jobs_and_queries = splunk.drilldown_enrichment(service, notable_data, 5) for i in range(len(jobs_and_queries)): job_and_queries = jobs_and_queries[i] assert job_and_queries[0] == "View all login attempts by system 'test_src'" assert ( job_and_queries[1] == '| from datamodel:"Authentication"."Authentication" | search src="\'test_src\'" | earliest=1d' ) assert isinstance(job_and_queries[2], client.Job) @pytest.mark.parametrize( "finding_data, expected_result", [ ( { "event_id": "test_id", "drilldown_name": "View all login attempts by system $src$", "drilldown_search": '| from datamodel:"Authentication"."Authentication" | search src=$src|s$', "drilldown_searches": '{"name":"View all login attempts by system $src$","search":"| from datamodel:\\"Authent' 'ication\\".\\"Authentication\\" | search src=$src|s$","earliest":1715040000,' '"latest":1715126400}', "_raw": "src='test_src'", "drilldown_latest": "1715126400.000000000", "drilldown_earliest": "1715040000.000000000", }, [ ( "View all login attempts by system 'test_src'", '| from datamodel:"Authentication"."Authentication" | search src="\'test_src\'"', ) ], ), ( { "event_id": "test_id2", "drilldown_searches": [ '{"name":"View all login attempts by system $src$","search":"| from datamodel:\\"Authentication\\".\\"Authe' 'ntication\\" | search src=$src|s$","earliest":1715040000,"latest":1715126400}', '{"name":"View all test involving user=\\"$user$\\"","search":"index=\\"test\\"\\n| where ' 'user = $user|s$","earliest":1716955500,"latest":1716959400}', ], "_raw": "src='test_src', user='test_user'", }, [ ( "View all login attempts by system 'test_src'", '| from datamodel:"Authentication"."Authentication" | search src="\'test_src\'"', ), ("View all test involving user=\"'test_user'\"", 'search index="test"\n| where user="\'test_user\'"'), ], ), ( { "event_id": "test_id3", "drilldown_searches": [ '{"name":"View all login attempts by system $src$","search":"| from datamodel:\\"Authentication\\".\\"Authe' 'ntication\\" | search src=$src|s$","earliest_offset":1715040000,"latest_offset":1715126400}', '{"name":"View all test involving user=\\"$user$\\"","search":"index=\\"test\\"\\n| where ' 'user = $user|s$","earliest_offset":1716955500,"latest_offset":1716959400}', ], "_raw": "src='test_src', user='test_user'", }, [ ( "View all login attempts by system 'test_src'", '| from datamodel:"Authentication"."Authentication" | search src="\'test_src\'"', ), ("View all test involving user=\"'test_user'\"", 'search index="test"\n| where user="\'test_user\'"'), ], ), ], ids=[ "A finding data with one drilldown search enrichment", "A finding data with two drilldown searches which contained the earlies in 'earliest' key ", "A finding data with two drilldown searches which contained the earlies in 'earliest_offset' key ", ], ) def test_drilldown_enrichment(finding_data, expected_result): """ Tests the logic of the drilldown_enrichment function. Given: 1. A finding data with one drilldown search enrichment. 2. A finding data with multiple (two) drilldown searches to enrich. When: - Running the splunk.drilldown_enrichment function. Then: - Verify that the returned jobs and queries are as expected. """ from splunklib import client service = Service("DONE") jobs_and_queries = splunk.drilldown_enrichment(service, finding_data, 5) for i in range(len(jobs_and_queries)): job_and_queries = jobs_and_queries[i] assert job_and_queries[0] == expected_result[i][0] assert job_and_queries[1] == expected_result[i][1] assert isinstance(job_and_queries[2], client.Job) @pytest.mark.parametrize( "finding_data, debug_log_message", [ ({"event_id": "test_id"}, "drill-down was not properly configured for finding test_id"), ( { "event_id": "test_id", "drilldown_name": "View all login attempts by system $src$", "drilldown_search": '| from datamodel:"Authentication"."Authentication" | search src=$src|s$', "_raw": "src='test_src'", "drilldown_latest": "", "drilldown_earliest": "", }, "Failed getting the drilldown timeframe for finding test_id", ), ( { "event_id": "test_id", "drilldown_name": "View all login attempts by system $src$", "drilldown_search": '| from datamodel:"Authentication"."Authentication" | search src=$src|s$', "_raw": "", "drilldown_latest": "00101", "drilldown_earliest": "00001", }, "Couldn't build search query for finding test_id with the following drilldown search ", ), ( { "event_id": "test_id", "drilldown_searches": [ '{"name":"View all login attempts by system $src$","search":"| from datamodel:\\"Authentica' 'tion\\".\\"Authentication\\" | search src=$src|s$","earliest":"","latest":""}', '{"name":"View all test involving user=\\"$user$\\"","search":"index=\\"test\\"\\n| where user =' '$user|s$","earliest":"","latest":""}', ], "_raw": "src='test_src', user='test_user'", }, "Failed getting the drilldown timeframe for finding test_id", ), ( { "event_id": "test_id", "drilldown_searches": [ '{"name":"View all login attempts by system $src$","search":"| from datamodel:\\"Authentic' 'ation\\".\\"Authentication\\" | search src=$src|s$","earliest":"","latest":""}', '{"name":"View all test involving user=\\"$user$\\"","search":"index=\\"test\\"\\n| where user =' '$user|s$","earliest":"","latest":""}', ], "_raw": "", }, "Couldn't build search query for finding test_id with the following drilldown search", ), ], ids=[ "A finding data without drilldown enrichment data", "A finding data with a single drilldown enrichment without search timeframe data", "A finding data with a single drilldown enrichment with an invalid search query", "A finding data with multiple drilldown enrichments without search timeframe data", "A finding data with multiple drilldown enrichments with invalid search queries", ], ) def test_drilldown_enrichment_no_enrichement_cases(mocker, finding_data, debug_log_message): """ Tests the logic of the drilldown_enrichment function when for some reason the enrichments raw data is invalid. Given: 1. A finding data without drilldown enrichment data. 2. A finding data with a single drilldown enrichment without search timeframe data. 3. A finding data with a single drilldown enrichment with an invalid search query. 4. A finding data with multiple drilldown enrichments without search timeframe data. 5. A finding data with multiple drilldown enrichments with invalid search queries. When: - Running the splunk.drilldown_enrichment function. Then: - Verify that the returned value is a tuple of None values as expected. """ debug_log = mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "error") service = Service("DONE") jobs_and_queries = splunk.drilldown_enrichment(service, finding_data, 5) for i in range(len(jobs_and_queries)): assert jobs_and_queries[i] == (None, None, None) assert debug_log_message in debug_log.call_args.args[0] """ ========== Mirroring Mechanism Tests ========== """ @pytest.mark.parametrize( "finding_data, func_call_kwargs, expected_closure_data", [ # A Finding with a "Closed" status label ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Closed", "event_id": "id", "rule_id": "id", "status_end": "true", "review_time": "1737547610.49", }, ], { "close_incident": True, "close_end_statuses": False, "close_extra_labels": [], }, { "EntryContext": {"mirrorRemoteId": "id"}, "Type": EntryType.NOTE, "Contents": { "dbotIncidentClose": True, "closeReason": 'Splunk event was closed on Splunk with status "Closed".', }, "ContentsFormat": EntryFormat.JSON, }, ), # A Finding with a "New" status label (shouldn't close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "New", "event_id": "id", "rule_id": "id", "status_end": "false", "review_time": "1737547610.50", }, ], { "close_incident": True, "close_end_statuses": False, "close_extra_labels": [], }, None, ), # A Finding with a custom status label that is on close_extra_labels (should close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Custom", "event_id": "id", "rule_id": "id", "status_end": "false", "review_time": "1737547610.51", }, ], { "close_incident": True, "close_end_statuses": False, "close_extra_labels": ["Custom"], }, { "EntryContext": {"mirrorRemoteId": "id"}, "Type": EntryType.NOTE, "Contents": { "dbotIncidentClose": True, "closeReason": 'Splunk event was closed on Splunk with status "Custom".', }, "ContentsFormat": EntryFormat.JSON, }, ), # A Finding with close_extra_labels that don't include status_label (shouldn't close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Custom", "event_id": "id", "rule_id": "id", "status_end": "false", "review_time": "1737547610.52", }, ], { "close_incident": True, "close_end_statuses": False, "close_extra_labels": ["A", "B"], }, None, ), # A Finding that has status_end as true with close_end_statuses as true (should close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Custom", "event_id": "id", "rule_id": "id", "status_end": "true", "review_time": "1737547610.53", }, ], { "close_incident": True, "close_end_statuses": True, "close_extra_labels": [], }, { "EntryContext": {"mirrorRemoteId": "id"}, "Type": EntryType.NOTE, "Contents": { "dbotIncidentClose": True, "closeReason": 'Splunk event was closed on Splunk with status "Custom".', }, "ContentsFormat": EntryFormat.JSON, }, ), # A Finding that has status_end as true with close_end_statuses as false (shouldn't close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Custom", "event_id": "id", "rule_id": "id", "status_end": "true", "review_time": "1737547610.54", }, ], { "close_incident": True, "close_end_statuses": False, "close_extra_labels": [], }, None, ), # A Finding that is both on close_extra_labels, # and has status_end as true with close_end_statuses as true (should close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Custom", "event_id": "id", "rule_id": "id", "status_end": "true", "review_time": "1737547610.55", }, ], { "close_incident": True, "close_end_statuses": True, "close_extra_labels": ["Custom"], }, { "EntryContext": {"mirrorRemoteId": "id"}, "Type": EntryType.NOTE, "Contents": { "dbotIncidentClose": True, "closeReason": 'Splunk event was closed on Splunk with status "Custom".', }, "ContentsFormat": EntryFormat.JSON, }, ), ], ) def test_get_modified_remote_data_command_close_incident( mocker, finding_data: list[results.Message | dict], func_call_kwargs: dict, expected_closure_data: dict ): class Jobs: def oneshot(self, query, **kwargs): assert kwargs["output_mode"] == splunk.OUTPUT_MODE_JSON return finding_data class Service: def __init__(self): self.jobs = Jobs() # Stand-in for service.kvstore — see _MagicKVStore docstring at module top. self.kvstore = _MagicKVStore() expected_entries = {"EntryContext": {"mirrorRemoteId": "id"}, "Type": EntryType.NOTE, "ContentsFormat": EntryFormat.JSON} args = {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"} mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mocker.patch.object(demisto, "params", return_value={"timezone": "0"}) mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "info") mocker.patch("SplunkPyV2.results.JSONResultsReader", return_value=finding_data) mocker.patch.object(demisto, "results") service = Service() splunk.get_modified_remote_data_command( service, args, mapper=splunk.UserMappingObject(service, False), **func_call_kwargs, ) results = demisto.results.call_args[0][0] expected_entries["Contents"] = finding_data[1] expected_results = [expected_entries] if expected_closure_data: expected_results.append(expected_closure_data) assert demisto.results.call_count == 1 assert results == expected_results def test_get_remote_data_command_with_message(mocker): """ Test for the get_remote_data_command function with a message. This test verifies that when the splunk-sdk returns a message, the function correctly logs the message using demisto.info(). Args: mocker: The mocker object for patching and mocking. Returns: None """ service = mocker.patch.object(client, "Service") mocker.patch.object(demisto, "info") mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": True, "close_end_statuses": True, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(service, False), } mocker.patch("SplunkPyV2.results.JSONResultsReader", return_value=[results.Message("INFO-test", "test message")]) splunk.get_modified_remote_data_command(service, **func_call_kwargs) assert demisto.info.call_args[0][0] == "Splunk-SDK message: test message" def test_fetch_with_error_in_message(mocker): """ Given - fetch result from Splunk return Error message When - fetch incidents Then - assert DemistoException is raised """ mock_params = {"fetchQuery": "something", "parseFindingEventsRaw": True} mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mocker.patch.object(splunk.client.Job, "is_done", return_value=True) mocker.patch("demistomock.getLastRun", return_value={"time": "2018-10-24T14:13:20"}) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=[results.Message("FATAL", "Error")]) # Phase 3b: dispatcher logs handler exceptions via demisto.error before # re-raising. Mute it here to satisfy the no-stdout fixture in conftest. mocker.patch.object(demisto, "error") # run service = mocker.patch("splunklib.client.connect") with pytest.raises(DemistoException) as e: splunk.fetch_incidents(service, None) assert "Failed to fetch incidents, check the provided query in Splunk web search" in e.value.message def test_get_modified_remote_data_command(mocker): updated_incidet_review = { "rule_id": "id", "event_id": "id", "review_time": "1737547610.56", } mocker.patch("SplunkPyV2.get_current_splunk_time", return_value="2021-02-09T17:41:30.589575+02:00") service = mocker.patch.object(client, "Service") func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": True, "close_end_statuses": True, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(service, False), } mocker.patch.object(demisto, "params", return_value={"timezone": "0"}) mocker.patch("SplunkPyV2.results.JSONResultsReader", return_value=[updated_incidet_review]) mocker.patch.object(demisto, "results") splunk.get_modified_remote_data_command(service, **func_call_kwargs) results = demisto.results.call_args[0][0][0]["Contents"] assert demisto.results.call_count == 1 assert results == updated_incidet_review def test_edit_finding_event__failed_to_update(mocker): """ Given - finding event with id ID100 When - updating the event with invalid owner 'dbot' - the service should return error string message 'ValueError: Invalid owner value.' Then - ensure the error message parsed correctly and returned to the user """ test_args = {"event_ids": "ID100", "owner": "dbot"} mocker.patch.object(demisto, "error") mocker.patch.object( splunk, "return_error", side_effect=Exception("Failed to update Splunk ES event ID100: ValueError: Invalid owner value.") ) mocker.patch.object(splunk, "update_investigation_or_finding", side_effect=Exception("ValueError: Invalid owner value.")) with pytest.raises(Exception, match="Failed to update Splunk ES event ID100: ValueError: Invalid owner value."): splunk.splunk_edit_event_command(service=MagicMock(), args=test_args) assert splunk.return_error.call_count == 1 error_message = splunk.return_error.call_args[0][0] assert "Failed to update Splunk ES event ID100: ValueError: Invalid owner value." in error_message NOTABLE = { "rule_name": "string", "rule_title": "string", "security_domain": "string", "index": "string", "rule_description": "string", "risk_score": "string", "host": "string", "host_risk_object_type": "string", "dest_risk_object_type": "string", "dest_risk_score": "string", "splunk_server": "string", "_sourcetype": "string", "_indextime": "string", "_time": "string", "src_risk_object_type": "string", "src_risk_score": "string", "_raw": "string", "urgency": "string", "owner": "string", "info_min_time": "string", "info_max_time": "string", "note": "string", "reviewer": "string", "rule_id": "string", "action": "string", "app": "string", "authentication_method": "string", "authentication_service": "string", "bugtraq": "string", "bytes": "string", "bytes_in": "string", "bytes_out": "string", "category": "string", "cert": "string", "change": "string", "change_type": "string", "command": "string", "comments": "string", "cookie": "string", "creation_time": "string", "cve": "string", "cvss": "string", "date": "string", "description": "string", "dest": "string", "dest_bunit": "string", "dest_category": "string", "dest_dns": "string", "dest_interface": "string", "dest_ip": "string", "dest_ip_range": "string", "dest_mac": "string", "dest_nt_domain": "string", "dest_nt_host": "string", "dest_port": "string", "dest_priority": "string", "dest_translated_ip": "string", "dest_translated_port": "string", "dest_type": "string", "dest_zone": "string", "direction": "string", "dlp_type": "string", "dns": "string", "duration": "string", "dvc": "string", "dvc_bunit": "string", "dvc_category": "string", "dvc_ip": "string", "dvc_mac": "string", "dvc_priority": "string", "dvc_zone": "string", "file_hash": "string", "file_name": "string", "file_path": "string", "file_size": "string", "http_content_type": "string", "http_method": "string", "http_referrer": "string", "http_referrer_domain": "string", "http_user_agent": "string", "icmp_code": "string", "icmp_type": "string", "id": "string", "ids_type": "string", "incident": "string", "ip": "string", "mac": "string", "message_id": "string", "message_info": "string", "message_priority": "string", "message_type": "string", "mitre_technique_id": "string", "msft": "string", "mskb": "string", "name": "string", "orig_dest": "string", "orig_recipient": "string", "orig_src": "string", "os": "string", "packets": "string", "packets_in": "string", "packets_out": "string", "parent_process": "string", "parent_process_id": "string", "parent_process_name": "string", "parent_process_path": "string", "password": "string", "payload": "string", "payload_type": "string", "priority": "string", "problem": "string", "process": "string", "process_hash": "string", "process_id": "string", "process_name": "string", "process_path": "string", "product_version": "string", "protocol": "string", "protocol_version": "string", "query": "string", "query_count": "string", "query_type": "string", "reason": "string", "recipient": "string", "recipient_count": "string", "recipient_domain": "string", "recipient_status": "string", "record_type": "string", "registry_hive": "string", "registry_key_name": "string", "registry_path": "string", "registry_value_data": "string", "registry_value_name": "string", "registry_value_text": "string", "registry_value_type": "string", "request_sent_time": "string", "request_payload": "string", "request_payload_type": "string", "response_code": "string", "response_payload_type": "string", "response_received_time": "string", "response_time": "string", "result": "string", "return_addr": "string", "rule": "string", "rule_action": "string", "sender": "string", "service": "string", "service_hash": "string", "service_id": "string", "service_name": "string", "service_path": "string", "session_id": "string", "sessions": "string", "severity": "string", "severity_id": "string", "sid": "string", "signature": "string", "signature_id": "string", "signature_version": "string", "site": "string", "size": "string", "source": "string", "sourcetype": "string", "src": "string", "src_bunit": "string", "src_category": "string", "src_dns": "string", "src_interface": "string", "src_ip": "string", "src_ip_range": "string", "src_mac": "string", "src_nt_domain": "string", "src_nt_host": "string", "src_port": "string", "src_priority": "string", "src_translated_ip": "string", "src_translated_port": "string", "src_type": "string", "src_user": "string", "src_user_bunit": "string", "src_user_category": "string", "src_user_domain": "string", "src_user_id": "string", "src_user_priority": "string", "src_user_role": "string", "src_user_type": "string", "src_zone": "string", "state": "string", "status": "string", "status_code": "string", "status_description": "string", "subject": "string", "tag": "string", "ticket_id": "string", "time": "string", "time_submitted": "string", "transport": "string", "transport_dest_port": "string", "type": "string", "uri": "string", "uri_path": "string", "uri_query": "string", "url": "string", "url_domain": "string", "url_length": "string", "user": "string", "user_agent": "string", "user_bunit": "string", "user_category": "string", "user_id": "string", "user_priority": "string", "user_role": "string", "user_type": "string", "vendor_account": "string", "vendor_product": "string", "vlan": "string", "xdelay": "string", "xref": "string", } DRILLDOWN = { "Drilldown": { "action": "string", "app": "string", "authentication_method": "string", "authentication_service": "string", "bugtraq": "string", "bytes": "string", "bytes_in": "string", "bytes_out": "string", "category": "string", "cert": "string", "change": "string", "change_type": "string", "command": "string", "comments": "string", "cookie": "string", "creation_time": "string", "cve": "string", "cvss": "string", "date": "string", "description": "string", "dest": "string", "dest_bunit": "string", "dest_category": "string", "dest_dns": "string", "dest_interface": "string", "dest_ip": "string", "dest_ip_range": "string", "dest_mac": "string", "dest_nt_domain": "string", "dest_nt_host": "string", "dest_port": "string", "dest_priority": "string", "dest_translated_ip": "string", "dest_translated_port": "string", "dest_type": "string", "dest_zone": "string", "direction": "string", "dlp_type": "string", "dns": "string", "duration": "string", "dvc": "string", "dvc_bunit": "string", "dvc_category": "string", "dvc_ip": "string", "dvc_mac": "string", "dvc_priority": "string", "dvc_zone": "string", "file_hash": "string", "file_name": "string", "file_path": "string", "file_size": "string", "http_content_type": "string", "http_method": "string", "http_referrer": "string", "http_referrer_domain": "string", "http_user_agent": "string", "icmp_code": "string", "icmp_type": "string", "id": "string", "ids_type": "string", "incident": "string", "ip": "string", "mac": "string", "message_id": "string", "message_info": "string", "message_priority": "string", "message_type": "string", "mitre_technique_id": "string", "msft": "string", "mskb": "string", "name": "string", "orig_dest": "string", "orig_recipient": "string", "orig_src": "string", "os": "string", "packets": "string", "packets_in": "string", "packets_out": "string", "parent_process": "string", "parent_process_id": "string", "parent_process_name": "string", "parent_process_path": "string", "password": "string", "payload": "string", "payload_type": "string", "priority": "string", "problem": "string", "process": "string", "process_hash": "string", "process_id": "string", "process_name": "string", "process_path": "string", "product_version": "string", "protocol": "string", "protocol_version": "string", "query": "string", "query_count": "string", "query_type": "string", "reason": "string", "recipient": "string", "recipient_count": "string", "recipient_domain": "string", "recipient_status": "string", "record_type": "string", "registry_hive": "string", "registry_key_name": "string", "registry_path": "string", "registry_value_data": "string", "registry_value_name": "string", "registry_value_text": "string", "registry_value_type": "string", "request_payload": "string", "request_payload_type": "string", "request_sent_time": "string", "response_code": "string", "response_payload_type": "string", "response_received_time": "string", "response_time": "string", "result": "string", "return_addr": "string", "rule": "string", "rule_action": "string", "sender": "string", "service": "string", "service_hash": "string", "service_id": "string", "service_name": "string", "service_path": "string", "session_id": "string", "sessions": "string", "severity": "string", "severity_id": "string", "sid": "string", "signature": "string", "signature_id": "string", "signature_version": "string", "site": "string", "size": "string", "source": "string", "sourcetype": "string", "src": "string", "src_bunit": "string", "src_category": "string", "src_dns": "string", "src_interface": "string", "src_ip": "string", "src_ip_range": "string", "src_mac": "string", "src_nt_domain": "string", "src_nt_host": "string", "src_port": "string", "src_priority": "string", "src_translated_ip": "string", "src_translated_port": "string", "src_type": "string", "src_user": "string", "src_user_bunit": "string", "src_user_category": "string", "src_user_domain": "string", "src_user_id": "string", "src_user_priority": "string", "src_user_role": "string", "src_user_type": "string", "src_zone": "string", "state": "string", "status": "string", "status_code": "string", "subject": "string", "tag": "string", "ticket_id": "string", "time": "string", "time_submitted": "string", "transport": "string", "transport_dest_port": "string", "type": "string", "uri": "string", "uri_path": "string", "uri_query": "string", "url": "string", "url_domain": "string", "url_length": "string", "user": "string", "user_agent": "string", "user_bunit": "string", "user_category": "string", "user_id": "string", "user_priority": "string", "user_role": "string", "user_type": "string", "vendor_account": "string", "vendor_product": "string", "vlan": "string", "xdelay": "string", "xref": "string", } } ASSET = { "Asset": { "asset": "string", "asset_id": "string", "asset_tag": "string", "bunit": "string", "category": "string", "city": "string", "country": "string", "dns": "string", "ip": "string", "is_expected": "string", "lat": "string", "long": "string", "mac": "string", "nt_host": "string", "owner": "string", "pci_domain": "string", "priority": "string", "requires_av": "string", } } IDENTITY = { "Identity": { "bunit": "string", "category": "string", "email": "string", "endDate": "string", "first": "string", "identity": "string", "identity_tag": "string", "last": "string", "managedBy": "string", "nick": "string", "phone": "string", "prefix": "string", "priority": "string", "startDate": "string", "suffix": "string", "watchlist": "string", "work_city": "string", "work_lat": "string", "work_long": "string", } } def test_get_cim_mapping_field_command(mocker): """Scenario: When the mapping is based on Splunk CIM.""" fields = splunk.get_cim_mapping_field_command() assert fields == {"Finding Data": NOTABLE, "Drilldown Data": DRILLDOWN, "Asset Data": ASSET, "Identity Data": IDENTITY} def test_build_search_human_readable(mocker): """ Given: table headers in query When: building a human readable table as part of splunk-search Then: Test headers are calculated correctly: * comma-separated, space-separated * support commas and spaces inside header values (if surrounded with parenthesis) * rename headers """ func_patch = mocker.patch("SplunkPyV2.update_headers_from_field_names") results = [ { "ID": 1, "Header with space": "h1", "header3": 1, "header_without_space": "1234", "old_header_1": "1", "old_header_2": "2", }, { "ID": 2, "Header with space": "h2", "header3": 2, "header_without_space": "1234", "old_header_1": "1", "old_header_2": "2", }, ] args = { "query": 'something | table ID "Header with space" header3 header_without_space ' 'comma,separated "Single,Header,with,Commas" old_header_1 old_header_2 | something else' " | rename old_header_1 AS new_header_1 old_header_2 AS new_header_2" } expected_headers = [ "ID", "Header with space", "header3", "header_without_space", "comma", "separated", "Single,Header,with,Commas", "new_header_1", "new_header_2", ] splunk.build_search_human_readable(args, results, sid="123456") headers = func_patch.call_args[0][1] assert headers == expected_headers def test_build_search_human_readable_multi_table_in_query(mocker): """ Given: multiple table headers in query When: building a human readable table as part of splunk-search Then: Test headers are calculated correctly: * all expected header exist without duplications """ args = {"query": " table header_1, header_2 | stats state_1, state_2 | table header_1, header_2, header_3, header_4"} results = [ {"header_1": "val_1", "header_2": "val_2", "header_3": "val_3", "header_4": "val_4"}, ] expected_headers_hr = "|header_1|header_2|header_3|header_4|\n|---|---|---|---|" hr = splunk.build_search_human_readable(args, results, sid="123456") assert expected_headers_hr in hr @pytest.mark.parametrize("polling, fast_mode", [(False, True), (True, True)]) def test_build_search_kwargs(polling, fast_mode): """ Given: The splunk-search command args. When: Running the build_search_kwargs to build the search query kwargs. Then: Ensure the query kwargs as expected. """ args = { "earliest_time": "2021-11-23T10:10:10", "latest_time": "2021-11-23T10:10:20", "app": "test_app", "fast_mode": fast_mode, "polling": polling, } kwargs_normalsearch = splunk.build_search_kwargs(args, polling) for field in args: if field == "polling": assert "exec_mode" in kwargs_normalsearch if polling: assert kwargs_normalsearch["exec_mode"] == "normal" else: assert kwargs_normalsearch["exec_mode"] == "blocking" elif field == "fast_mode" and fast_mode: assert kwargs_normalsearch["adhoc_search_level"] == "fast" else: assert field in kwargs_normalsearch @pytest.mark.parametrize("polling,status", [(False, "DONE"), (True, "DONE"), (True, "RUNNING")]) def test_splunk_search_command(mocker, polling, status): """ Given: A search query with args. When: Running the splunk_search_command with and without polling. Then: Ensure the result as expected in polling and in regular search. """ mock_args = { "query": "query", "earliest_time": "2021-11-23T10:10:10", "latest_time": "2020-10-20T10:10:20", "app": "test_app", "fast_mode": "false", "polling": polling, } mocker.patch.object(ScheduledCommand, "raise_error_if_not_supported") search_result = splunk.splunk_search_command(Service(status), mock_args) search_result = search_result if isinstance(search_result, CommandResults) else search_result[0] if search_result.scheduled_command: assert search_result.outputs["Status"] == status assert search_result.scheduled_command._args["sid"] == "123456" else: assert search_result.outputs["Splunk.Result"] == [] assert search_result.readable_output == "### Splunk Search results for query:\nsid: 123456\n**No entries.**\n" @pytest.mark.parametrize( "messages,expected_msg", [({"fatal": ["fatal msg"]}, "fatal msg"), ({"error": ["error msg"]}, "error msg")] ) def test_err_in_splunk_search(mocker, messages, expected_msg): """ Given: A wrong search query. When: Running the splunk_search_command. Then: Ensure the result as expected in polling and in regular search. """ mock_args = { "query": "wrong search query", "earliest_time": "2021-11-23T10:10:10", "latest_time": "2020-10-20T10:10:20", "fast_mode": "false", } service = Service(status="FAILED") service.jobs.state.content["messages"] = messages with pytest.raises(DemistoException) as e: splunk.splunk_search_command(service, mock_args) assert f"Failed to run the search in Splunk: {expected_msg}" in str(e) @pytest.mark.parametrize( argnames="credentials", argvalues=[{"username": "test", "password": "test"}, {"splunkToken": "token", "password": "test"}] ) def test_module_test(mocker, credentials): """ Given: - Credentials for connecting Splunk When: - Run test-module command Then: - Validate the info method was called """ # prepare mocker.patch.object(client.Service, "info") mocker.patch.object(client.Service, "login") service = client.Service(**credentials) # run splunk.test_module(service, {}) # validate assert service.info.call_count == 1 @pytest.mark.parametrize( argnames="credentials", argvalues=[{"username": "test", "password": "test"}, {"splunkToken": "token", "password": "test"}] ) def test_module__exception_raised(mocker, credentials): """ Given: - AuthenticationError was occurred When: - Run test-module command Then: - Validate the expected message was returned """ # prepare def exception_raiser(): raise AuthenticationError mocker.patch.object(AuthenticationError, "__init__", return_value=None) mocker.patch.object(client.Service, "info", side_effect=exception_raiser) mocker.patch.object(client.Service, "login") return_error_mock = mocker.patch(RETURN_ERROR_TARGET) service = client.Service(**credentials) # run splunk.test_module(service, {}) # validate assert return_error_mock.call_args[0][0] == "Authentication error, please validate your credentials." def test_module_hec_url(mocker): """ Given: - hec_url was is in params When: - Run test-module command Then: - Validate that the request.get was called with the expected args """ # prepare mocker.patch.object(client.Service, "info") mocker.patch.object(client.Service, "login") mocker.patch.object(requests, "get") service = client.Service(username="test", password="test") # run splunk.test_module(service, {"hec_url": "test_hec_url"}) # validate assert requests.get.call_args[0][0] == "test_hec_url/services/collector/health" def test_module_message_object(mocker): """ Given: - query results with one message item. When: - Run test-module command. Then: - Validate the test_module run successfully and the info method was called once. """ # prepare message = results.Message("DEBUG", "There's something in that variable...") mocker.patch("splunklib.results.JSONResultsReader", return_value=[message]) service = mocker.patch("splunklib.client.connect", return_value=None) # run splunk.test_module(service, {"isFetch": True, "fetchQuery": "something"}) # validate assert service.info.call_count == 1 def test_module_investigations_query_missing_placeholder_raises(mocker): """ Given: - isFetch is enabled - "Investigation" is selected in `fetch_event_types` - `investigations_fetch_query` does NOT contain FETCH_FILTER_PLACEHOLDER When: - Run test-module command Then: - DemistoException is raised whose message names the missing token and the parameter to fix """ # prepare message = results.Message("DEBUG", "stub") mocker.patch("splunklib.results.JSONResultsReader", return_value=[message]) service = mocker.patch("splunklib.client.connect", return_value=None) params = { "isFetch": True, "fetchQuery": "search `notable`", "fetch_event_types": ["Finding", "Investigation"], "investigations_fetch_query": '| rest "/servicesNS/nobody/missioncontrol/public/v2/investigations?search_format=true"', } # run + validate with pytest.raises(splunk.DemistoException) as exc_info: splunk.test_module(service, params) msg = str(exc_info.value) assert "FETCH_FILTER_PLACEHOLDER" in msg assert "Investigations fetch query" in msg def test_module_investigations_not_selected_skips_validation(mocker): """ Given: - isFetch is enabled - "Investigation" is NOT in `fetch_event_types` - `investigations_fetch_query` is set to an invalid value (no placeholder) When: - Run test-module command Then: - test_module completes without raising the placeholder validation error (the invalid investigations query is irrelevant when Investigation isn't selected) """ # prepare message = results.Message("DEBUG", "stub") mocker.patch("splunklib.results.JSONResultsReader", return_value=[message]) service = mocker.patch("splunklib.client.connect", return_value=None) params = { "isFetch": True, "fetchQuery": "search `notable`", "fetch_event_types": ["Finding"], # Investigation NOT selected # Intentionally invalid query — must not be probed when Investigation isn't selected. "investigations_fetch_query": '| rest "/path/v2/investigations?no_placeholder_here"', } # run — must not raise the placeholder DemistoException splunk.test_module(service, params) # validate connection still probed assert service.info.call_count == 1 def test_labels_with_non_str_values(mocker): """ Given: - Raw response with values in _raw that stored as dict or list When: - Fetch incidents Then: - Validate the Labels created in the incident are well formatted to avoid server errors on json.Unmarshal """ from SplunkPyV2 import UserMappingObject # prepare raw = { "message": "Authentication of user via Radius", "actor_obj": {"id": "test", "type": "User", "alternateId": "test", "displayName": "test"}, "actor_list": [{"id": "test", "type": "User", "alternateId": "test", "displayName": "test"}], "actor_tuple": ("id", "test"), "num_val": 100, "bool_val": False, "float_val": 100.0, } mocked_response: list[results.Message | dict] = deepcopy(SAMPLE_RESPONSE) mocked_response[1]["_raw"] = json.dumps(raw) mock_last_run = {"time": "2018-10-24T14:13:20"} mock_params = {"fetchQuery": "something", "parseFindingEventsRaw": True} mocker.patch.object(demisto, "incidents") mocker.patch.object(demisto, "setLastRun") mocker.patch("SplunkPyV2.get_current_splunk_time", return_value="2021-02-09T17:41:30.589575+02:00") mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=mocked_response) # run service = mocker.patch("splunklib.client.connect", return_value=None) mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper) incidents = demisto.incidents.call_args[0][0] # validate assert demisto.incidents.call_count == 1 assert len(incidents) == 2 labels = incidents[0]["labels"] assert len(labels) >= 7 assert all(isinstance(label["value"], str) for label in labels) def test_empty_string_as_app_param_value(mocker): """ Given: - A mock to demisto.params that contains an 'app' key with an empty string as its value When: - Run splunk.get_connection_args() function Then: - Validate that the value of the 'app' key in connection_args is '-' """ # prepare mock_params = {"app": "", "host": "111", "port": "111"} # run connection_args = splunk.get_connection_args(mock_params) # validate assert connection_args.get("app") == "-" OWNER_MAPPING = [ {"xsoar_user": "test_xsoar", "splunk_user": "test_splunk", "wait": True}, {"xsoar_user": "test_not_full", "splunk_user": "", "wait": True}, {"xsoar_user": "", "splunk_user": "test_not_full", "wait": True}, ] MAPPER_CASES_XSOAR_TO_SPLUNK = [ ( "", "unassigned", "UserMapping: Could not find splunk user matching xsoar's . Consider adding it to the splunk_xsoar_users lookup.", ), ( "not_in_table", "unassigned", "UserMapping: Could not find splunk user matching xsoar's not_in_table. " "Consider adding it to the splunk_xsoar_users lookup.", ), ] @pytest.mark.parametrize("xsoar_name, expected_splunk, expected_msg", MAPPER_CASES_XSOAR_TO_SPLUNK) def test_owner_mapping_mechanism_xsoar_to_splunk(mocker, xsoar_name, expected_splunk, expected_msg): """ Given: - different xsoar values When: - fetching, or mirroring Then: - validates the splunk user is correct """ def mocked_get_record(col, value_to_search): return filter(lambda x: x[col] == value_to_search, OWNER_MAPPING[:-1]) service = mocker.patch("splunklib.client.connect", return_value=None) mapper = splunk.UserMappingObject( service, True, table_name="splunk_xsoar_users", xsoar_user_column_name="xsoar_user", splunk_user_column_name="splunk_user" ) mocker.patch.object(mapper, "_get_record", side_effect=mocked_get_record) error_mock = mocker.patch.object(demisto, "error") s_user = mapper.get_splunk_user_by_xsoar(xsoar_name) assert s_user == expected_splunk if error_mock.called: assert error_mock.call_args[0][0] == expected_msg MAPPER_CASES_SPLUNK_TO_XSOAR = [ ("test_splunk", "test_xsoar", None), ( "test_not_full", "", "UserMapping: Xsoar user matching splunk's test_not_full is empty. Fix the record in splunk_xsoar_users lookup.", ), ( "unassigned", "", "UserMapping: Could not find xsoar user matching splunk's unassigned. Consider adding it to the" " splunk_xsoar_users lookup.", ), ( "not_in_table", "", "UserMapping: Could not find xsoar user matching splunk's not_in_table. " "Consider adding it to the splunk_xsoar_users lookup.", ), ] @pytest.mark.parametrize("splunk_name, expected_xsoar, expected_msg", MAPPER_CASES_SPLUNK_TO_XSOAR) def test_owner_mapping_mechanism_splunk_to_xsoar(mocker, splunk_name, expected_xsoar, expected_msg): """ Given: - different xsoar values When: - fetching, or mirroring Then: - validates the splunk user is correct """ def mocked_get_record(col, value_to_search): return filter(lambda x: x[col] == value_to_search, OWNER_MAPPING) service = mocker.patch("splunklib.client.connect", return_value=None) mapper = splunk.UserMappingObject( service, True, table_name="splunk_xsoar_users", xsoar_user_column_name="xsoar_user", splunk_user_column_name="splunk_user" ) mocker.patch.object(mapper, "_get_record", side_effect=mocked_get_record) error_mock = mocker.patch.object(demisto, "error") s_user = mapper.get_xsoar_user_by_splunk(splunk_name) assert s_user == expected_xsoar if error_mock.called: assert error_mock.call_args[0][0] == expected_msg COMMAND_CASES = [ ( {"xsoar_username": "test_xsoar"}, # case normal single username was provided [{"SplunkUser": "test_splunk", "XsoarUser": "test_xsoar"}], ), ( {"xsoar_username": "test_xsoar, Non existing"}, # case normal multiple usernames were provided [{"SplunkUser": "test_splunk", "XsoarUser": "test_xsoar"}, {"SplunkUser": "unassigned", "XsoarUser": "Non existing"}], ), ( {"xsoar_username": "Non Existing,"}, # case normal&empty multiple usernames were provided [ {"SplunkUser": "unassigned", "XsoarUser": "Non Existing"}, {"SplunkUser": "Could not map splunk user, Check logs for more info.", "XsoarUser": ""}, ], ), ( {"xsoar_username": ["test_xsoar", "Non existing"]}, # case normal&missing multiple usernames were provided [{"SplunkUser": "test_splunk", "XsoarUser": "test_xsoar"}, {"SplunkUser": "unassigned", "XsoarUser": "Non existing"}], ), ( {"xsoar_username": ["test_xsoar", "Non existing"], "map_missing": False}, # case normal & missing multiple usernames were provided without missing's mapping activated [ {"SplunkUser": "test_splunk", "XsoarUser": "test_xsoar"}, {"SplunkUser": "Could not map splunk user, Check logs for more info.", "XsoarUser": "Non existing"}, ], ), ( {"xsoar_username": "Non Existing,", "map_missing": False}, # case missing&empty multiple usernames were provided [ {"SplunkUser": "Could not map splunk user, Check logs for more info.", "XsoarUser": "Non Existing"}, {"SplunkUser": "Could not map splunk user, Check logs for more info.", "XsoarUser": ""}, ], ), ] @pytest.mark.parametrize("xsoar_names, expected_outputs", COMMAND_CASES) def test_get_splunk_user_by_xsoar_command(mocker, xsoar_names, expected_outputs): """ Given: a list of xsoar users When: trying to get splunk matching users Then: validates correctness of list """ def mocked_get_record(col, value_to_search): return filter(lambda x: x[col] == value_to_search, OWNER_MAPPING[:-1]) service = mocker.patch("splunklib.client.connect", return_value=None) mapper = splunk.UserMappingObject( service, True, table_name="splunk_xsoar_users", xsoar_user_column_name="xsoar_user", splunk_user_column_name="splunk_user" ) # Ignoring logging pytest error mocker.patch.object(demisto, "error") mocker.patch.object(mapper, "_get_record", side_effect=mocked_get_record) res = mapper.get_splunk_user_by_xsoar_command(xsoar_names) assert res.outputs == expected_outputs def test_authentication_params(mocker): """ Given: - the splunkToken When: - connecting to Splunk server Then: - validate the connection args was sent as expected """ splunk_token = "splunk_token" mocked_params = { "server_url": "test_host", "proxy": "false", "authentication": {"identifier": "identifier", "password": splunk_token}, } mocker.patch.object(client, "connect") mocker.patch.object(demisto, "params", return_value=mocked_params) mocker.patch.object(demisto, "command", return_value="unknown-command") with pytest.raises(NotImplementedError): splunk.main() assert client.connect.call_args[1]["splunkToken"] == splunk_token @pytest.mark.parametrize( argnames="host, expected_connect_args", argvalues=[ ("8.8.8.8", {"host": "8.8.8.8", "port": 8089}), ("https://www.test.com", {"host": "www.test.com", "port": 8089}), ("www.test.com:9000/", {"host": "www.test.com", "port": 9000}), ], ) def test_server_url(mocker, host, expected_connect_args): """ Given: - Different server url values When: - Running the test-module command Then: - Ensure the connection args sent as expected """ mocked_params = { "server_url": host, "proxy": "false", "authentication": {"identifier": "username", "password": "splunk_token"}, } mocker.patch.object(demisto, "command", return_value="test-module") mocker.patch.object(demisto, "params", return_value=mocked_params) mocked_connect = mocker.patch.object(client, "connect") mocker.patch.object(splunk, "test_module") splunk.main() assert all(mocked_connect.call_args[1][k] == expected_connect_args[k] for k in expected_connect_args) @pytest.mark.parametrize( "item, expected", [({"message": "Test message"}, False), (results.Message("INFO", "Test message"), True)] ) def test_handle_message(item: dict | results.Message, expected: bool): """ Tests that passing a results.Message object returns True """ assert splunk.handle_message(item) is expected def test_single_drilldown_searches(mocker): """ Given: - finding with single string represent dict, in the drilldown_searches key. When: - call to drilldown_enrichment. Then: - validate there is no errors in the process. """ drilldown_searches = json.dumps( {"name": "test drilldown", "search": "| from datamodel: test", "earliest": 1719218100, "latest": 1719823500} ) mocker.patch.object(demisto, "error") mocker.patch.object(splunk, "build_drilldown_search", return_value=None) splunk.drilldown_enrichment( service=None, finding_data={"drilldown_searches": drilldown_searches, "event_id": "test_id"}, num_enrichment_events=1 ) assert demisto.error.call_count == 0, "Something was wrong in the drilldown_enrichment process" @pytest.mark.parametrize( "drilldown_data, expected", [ ({"drilldown_search": "test"}, ["test"]), ({"drilldown_searches": '{"search_1":"test_1"}'}, [{"search_1": "test_1"}]), ( {"drilldown_searches": ['{"search_1":"test_1"}', '{"search_2":"test_2"}']}, [{"search_1": "test_1"}, {"search_2": "test_2"}], ), ({"drilldown_searches": '[{"search_1":"test_1"}]'}, [{"search_1": "test_1"}]), ( {"drilldown_searches": '[{"search_1":"test_1"}, {"search_2":"test_2"}]'}, [{"search_1": "test_1"}, {"search_2": "test_2"}], ), ], ) def test_get_drilldown_searches(drilldown_data, expected): """ Given: - 1. A finding data with a single 'old' (string value in the 'drilldown_search' key) drilldown enrichment data . 2. A finding data with a single drilldown enrichments as json string in the 'new' key (drilldown_searches). 3. A finding data with multiple drilldown enrichments as json string in the 'new' key (drilldown_searches). 4. A finding data with a single drilldown enrichments as json list string in the 'new' key (drilldown_searches). 5. A finding data with a multiple drilldown enrichments as json list string in the 'new' key (drilldown_searches). When: - call to get_drilldown_searches. Then: - validate the result are as expected. """ assert splunk.get_drilldown_searches(drilldown_data) == expected @pytest.mark.parametrize( "drilldown_search, expected_res", [ ('{"name":"test", "query":"|key="the value""}', 'key="the value"'), ('{"name":"test", "query":"|key in (line_1\nline_2)"}', "key in (line_1,line_2)"), ('{"name":"test", "query":"search a=$a|s$ c=$c$ suffix"}', "search a=$a|s$ c=$c$ suffix"), ], ) def test_escape_invalid_chars_in_drilldown_json(drilldown_search, expected_res): """ Scenario: When extracting the drilldown search query which are a json string, we should escape unescaped JSON special characters. Given: - A raw search query with text like 'key="a value"'. - A raw search query with text like where 'key in (a\nb)' which it should be 'key in (a,b)'. - A raw search query with normal json string, should not be changed by this function. When: - escape_invalid_chars_in_drilldown_json is called Then: - Return the expected result """ import json res = splunk.escape_invalid_chars_in_drilldown_json(drilldown_search) assert expected_res in json.loads(res)["query"] # Define minimal classes to simulate the service and index behavior class Index: def __init__(self, name): self.name = name class ServiceIndex: def __init__(self, indexes): self.indexes = [Index(name) for name in indexes] @pytest.mark.parametrize( "fields, expected", [ # Valid JSON input ('{"key": "value"}', {"key": "value"}), # Valid JSON with multiple key-value pairs ('{"key1": "value1", "key2": 2}', {"key1": "value1", "key2": 2}), # Invalid JSON input (non-JSON string) ("not a json string", {"fields": "not a json string"}), # Another invalid JSON input (partially structured JSON) ("{'key': 'value'}", {"fields": "{'key': 'value'}"}), ], ) def test_parse_fields(fields, expected): """ Given: A string representing fields, which may be a valid JSON string or a regular string. When: The parse_fields function is called with the given string. Then: If the string is valid JSON, the function returns a dictionary of the parsed fields. If the string is not valid JSON, the function returns a dictionary with a single key-value pair, where the entire input string is the key. """ from SplunkPyV2 import parse_fields result = parse_fields(fields) assert result == expected @pytest.mark.parametrize( "event, batch_event_data, entry_id, expected_data", [ ("Somthing happened", None, None, '{"event": "Somthing happened", "fields": {"field1": "value1"}, "index": "main"}'), ( None, "{'event': 'some event', 'index': 'some index'} {'event': 'some event', 'index': 'some index'}", None, "{'event': 'some event', 'index': 'some index'} {'event': 'some event', 'index': 'some index'}", ), # Batch event data ( None, None, "some entry_id", "{'event': 'some event', 'index': 'some index'} {'event': 'some event', 'index': 'some index'}", ), ( None, """{'event': "some event's", 'index': 'some index'} {'event': 'some event', 'index': 'some index'}""", None, """{'event': "some event's", 'index': 'some index'} {'event': 'some event', 'index': 'some index'}""", ), # with ' ( None, None, "some entry_id", "{'event': 'some event', 'index': 'some index'} {'event': 'some event', 'index': 'some index'}", ), ], ) @patch("requests.post") @patch("SplunkPyV2.get_events_from_file") @patch("SplunkPyV2.parse_fields") def test_splunk_submit_event_hec( mock_parse_fields, mock_get_events_from_file, mock_post, event, batch_event_data, entry_id, expected_data, ): """ Given: Different types of event submission (single event, batch event, entry_id). When: Calling splunk_submit_event_hec. Then: Ensure a POST request is sent with the correct data and headers. """ from SplunkPyV2 import splunk_submit_event_hec # Arrange hec_token = "valid_token" baseurl = "https://splunk.example.com" fields = '{"field1": "value1"}' parsed_fields = {"field1": "value1"} # Mocks mock_parse_fields.return_value = parsed_fields if entry_id: # Entry ID mock_get_events_from_file.return_value = ( "{'event': 'some event', 'index': 'some index'} {'event': 'some event', 'index': 'some index'}" ) # Act splunk_submit_event_hec( hec_token=hec_token, baseurl=baseurl, event=event, fields=fields, host=None, index="main", source_type=None, source=None, time_=None, request_channel="test_channel", batch_event_data=batch_event_data, entry_id=entry_id, service=MagicMock(), ) mock_post.assert_called_once_with( f"{baseurl}/services/collector/event", data=expected_data, headers={ "Authorization": f"Splunk {hec_token}", "Content-Type": "application/json", "X-Splunk-Request-Channel": "test_channel", }, verify=splunk.VERIFY_CERTIFICATE, ) def test_splunk_submit_event_hec_command_no_required_arguments(): """Given: none of these arguments: 'entry_id', 'event', 'batch_event_data' When: Runing splunk-submit-event-hec command Then: An exception is thrown """ from SplunkPyV2 import splunk_submit_event_hec_command with pytest.raises( DemistoException, match=r"Invalid input: Please specify one of the following arguments: `event`, `batch_event_data`, or `entry_id`.", ): splunk_submit_event_hec_command({"hec_url": "hec_url"}, None, {}) @pytest.mark.parametrize(argnames="should_map_user", argvalues=[True, False]) def test_get_modified_remote_data_command_with_user_mapping(mocker, should_map_user): """Given: - Different values for the splunk.UserMappingObject.should_map arguments and `notable` query response without 'owner' key When: - Runing test_get_modified_remote_data_command Then: - Verify the correct owner are returned. """ finding_without_owner = deepcopy(SAMPLE_RESPONSE[2]) del finding_without_owner["owner"] mapped_user = "mapped_splunk_user" mocker.patch.object(demisto, "results") mocker.patch.object(demisto, "params", return_value={"timezone": "0"}) mocker.patch.object(splunk.UserMappingObject, "get_xsoar_user_by_splunk", return_value=mapped_user) mocker.patch("SplunkPyV2.results.JSONResultsReader", side_effect=lambda res: res) mocked_service = mocker.patch("SplunkPyV2.client.Service") mocker.patch("SplunkPyV2.get_integration_context") mocked_service.jobs.oneshot = ( lambda query, **kwargs: [SAMPLE_AUDIT_INDEX_RESPONSE[0]] if "index=_audit" in query else [finding_without_owner] ) splunk.get_modified_remote_data_command( mocked_service, args={"lastUpdate": "2021-02-09T16:41:30.589575+02:00"}, mapper=splunk.UserMappingObject(mocked_service, should_map_user), close_incident=True, close_end_statuses=False, close_extra_labels=[], ) contents = demisto.results.call_args[0][0][0]["Contents"] expected_owner = mapped_user if should_map_user else SAMPLE_AUDIT_INDEX_RESPONSE[0]["owner"] assert contents["owner"] == expected_owner def test_mirror_in_with_enrichment_enabled(mocker): """ Given: - Drilldown Enrichmnet enabled in the instance configuration When: - Mirror in run (get-modified-remote-data) Then: - Validate the integration context stored the "delta" for the incident which sent to enrichment but not yet created in order to create the incident with the updated fields. """ # create an integration context in order to simulate the context in a normal run. integration_context = { splunk.CACHE: json.dumps( {splunk.SUBMITTED_FINDINGS: [splunk.Finding(SAMPLE_RESPONSE[2])]}, default=lambda obj: obj.__dict__ ), } mocker.patch("SplunkPyV2.set_integration_context") mocker.patch("SplunkPyV2.get_integration_context", return_value=integration_context) mocker.patch.object(demisto, "params", return_value={}) mocker.patch.object(splunk, "ENABLED_ENRICHMENTS", new=[splunk.DRILLDOWN_ENRICHMENT]) mocker.patch("SplunkPyV2.results.JSONResultsReader", side_effect=lambda res: res) mocker.patch.object(splunk.UserMappingObject, "get_xsoar_user_by_splunk", return_value="after_mirror_owner") mocker.patch("SplunkPyV2.get_current_splunk_time", return_value="2021-02-09T17:41:30.589575+02:00") mocked_service = mocker.patch("SplunkPyV2.client.Service") finding_delta = {"status_label": "after_mirror_status", "urgency": "after_mirror_urgency"} updated_finding = SAMPLE_RESPONSE[2] | finding_delta mocked_service.jobs.oneshot.return_value = [updated_finding] splunk.get_modified_remote_data_command( mocked_service, args={"lastUpdate": "2021-02-09T16:41:30.589575+02:00"}, mapper=splunk.UserMappingObject(mocked_service, True), close_incident=True, close_end_statuses=False, close_extra_labels=[], ) finding_id = SAMPLE_RESPONSE[2]["event_id"] mirrored_enriching_natables = splunk.set_integration_context.call_args[0][0][splunk.MIRRORED_ENRICHING_FINDINGS] actual_mirrored_finding_delta = mirrored_enriching_natables[finding_id] assert actual_mirrored_finding_delta["owner"] == "after_mirror_owner" assert all(actual_mirrored_finding_delta[k] == v for k, v in finding_delta.items()) def test_format_splunk_note_for_xsoar_basic(): """ Given: - A Splunk note with URL-encoded title and content. When: - Formatting the note for XSOAR. Then: - The title and content are URL-decoded and separated by a blank line, ending with a newline. """ note = {"title": "My%20Title", "content": "Line%201%0ALine%202", "author": {"username": "test user"}} expected = "**test user**\n\nMy Title\nLine 1\nLine 2" assert splunk.format_splunk_note_for_xsoar(note) == expected def test_format_splunk_note_for_xsoar_empty_content(): """ Given: - A Splunk note with a URL-encoded title and empty content. When: - Formatting the note for XSOAR. Then: - The title is URL-decoded and followed by two newlines (current implementation behavior). """ note = {"title": "Only%20Title", "content": "", "author": {"username": "test user"}} expected = "**test user**\n\nOnly Title" assert splunk.format_splunk_note_for_xsoar(note) == expected def test_user_mapping_used_cache(mocker): """ Given: - A KVStore table exist in SPlunk to map the Splunk user to the XSOAR user. When: - Call to the function to map the user. Then: - Validate that the function use cache to store the mapped values and called only once. """ mocker.patch.object(demisto, "error") mocked_service = mocker.patch("SplunkPyV2.client.Service") mapper = splunk.UserMappingObject(mocked_service, True) for _ in range(5): mapper.get_xsoar_user_by_splunk("test_splunk_user") assert mocked_service.kvstore.__getitem__().data.query.call_count == 1 @pytest.mark.parametrize( "query, expected_query", [ ("search index=_internal", "search index=_internal"), ("| inputlookup some_lookup", "| inputlookup some_lookup"), ("index=_internal", "search index=_internal"), ], ) def test_splunk_job_create_command(mocker, query, expected_query): mocked_service = mocker.patch("SplunkPyV2.client.Service") mocked_create_job = MagicMock() mocked_service.jobs.create = mocked_create_job mocker.patch("SplunkPyV2.return_results") args = {"query": query} splunk.splunk_job_create_command(mocked_service, args) mocked_create_job.assert_called_once_with(expected_query, exec_mode="normal", app="") def mock_service_job(sid): class MockJob: def __init__(self, state): self.state = MagicMock() self.state.content = {"dispatchState": state} class MockResponse: def __init__(self, status, reason, body): self.status = status self.reason = reason self.body = body self.headers = {} class MockBody: def __init__(self, message): self.message = message def read(self): return self.message if sid == "valid_sid": return MockJob("DONE") elif sid == "running_sid": return MockJob("RUNNING") elif sid == "error_sid": raise HTTPError(MockResponse("418", "I'm a teapot", MockBody("I won't brew coffee."))) else: raise HTTPError(MockResponse("404", "Not Found", MockBody("Unknown sid."))) @patch("SplunkPyV2.client.Service") def test_splunk_job_status_valid(mock_service): mock_service.job.side_effect = mock_service_job service = mock_service args = {"sid": "valid_sid"} result = splunk.splunk_job_status(service, args) assert len(result) == 1 assert result[0].outputs == {"SID": "valid_sid", "Status": "DONE"} assert "Splunk Job Status" in result[0].readable_output @patch("SplunkPyV2.client.Service") def test_splunk_job_status_running(mock_service): mock_service.job.side_effect = mock_service_job service = mock_service args = {"sid": "running_sid"} result = splunk.splunk_job_status(service, args) assert len(result) == 1 assert result[0].outputs == {"SID": "running_sid", "Status": "RUNNING"} assert "Splunk Job Status" in result[0].readable_output @patch("SplunkPyV2.client.Service") def test_splunk_job_status_not_found(mock_service): mock_service.job.side_effect = mock_service_job service = mock_service args = {"sid": "invalid_sid"} result = splunk.splunk_job_status(service, args) assert len(result) == 1 assert result[0].readable_output == "Not found job for SID: invalid_sid" @patch("SplunkPyV2.client.Service") def test_splunk_job_status_418_error(mock_service): mock_service.job.side_effect = mock_service_job service = mock_service args = {"sid": "error_sid"} result = splunk.splunk_job_status(service, args) assert len(result) == 1 assert ( "Querying splunk for SID: error_sid resulted in the following error HTTP 418 I'm a teapot -- I won't brew coffee" in result[0].readable_output ) @patch("SplunkPyV2.client.Service") def test_splunk_job_status_multiple_sids(mock_service): mock_service.job.side_effect = mock_service_job service = mock_service args = {"sid": "valid_sid,running_sid,invalid_sid"} result = splunk.splunk_job_status(service, args) assert len(result) == 3 assert result[0].outputs == {"SID": "valid_sid", "Status": "DONE"} assert result[1].outputs == {"SID": "running_sid", "Status": "RUNNING"} assert result[2].readable_output == "Not found job for SID: invalid_sid" def test_splunk_search_parse_bad_chars(): """ Given: The splunk search output contains a json string with invalid chars. (e.g. 0xa0, 0xd1 etc.) When: Attempting to parse the results from splunk search. Then: The parsing removes the bad chars and proceeds successfully. """ import io bad_search_output = b'{"preview": false, "init_offset": 0, "messages": [], "fields": [{"name": "Message"}, {"name": "_bkt"}, \ {"name": "_cd"}, {"name": "_indextime"}, {"name": "_pre_msg"}, {"name": "_raw"}, {"name": "_serial"}, {"name": "_si"}, \ {"name": "_sourcetype"}, {"name": "_time"}, {"name": "host"}, {"name": "index"}, {"name": "linecount"}, \ {"name": "source"}, {"name": "sourcetype"}, {"name": "splunk_server"}], \ "results": [{"Message": "Service \xd1started\xa0 successfully.", "_bkt": "main~1111~00000000-0000-0000-0000-000000000000", \ "_cd": "1111:0000000", "_indextime": "5555555555", "_pre_msg": "04/23/2025 08:04:41 AM\\nLogName=Test log\\n\ SourceName=Server\\nEventCode=0\\nEventType=4\\nType=Information\xa0\\nComputerName=#COMPUTERNAME#\\nTaskCategory=\ Test log Server\\nOpCode=Info\\nRecordNumber=3\\nKeywords=Classic", "_raw": "04/23/2025 08:04:41 AM\\nLogName=Test log\\n\ SourceName=Server\\nEventCode=0\\nEventType=4\\nType=Information\xa0\\nComputerName=#COMPUTERNAME#\\nTaskCategory=Test log \ Server\\nOpCode=Info\\nRecordNumber=3\\nKeywords=Classic\\nMessage=Service started successfully.\\n", "_serial": "1", \ "_si": ["ip-000-00-00-000", "main"], "_sourcetype": "WinEventLog", "_time": "2025-04-23T05:04:41.000-03:00", \ "host": "127.0.0.1", "index": "main", "linecount": "13", "source": "WinEventLog:Server", "sourcetype": "WinEventLog", \ "splunk_server": "ip-000-00-00-000"}], "highlighted": {}}' expected_res = ( [ { "Message": "Service started successfully.", "_bkt": "main~1111~00000000-0000-0000-0000-000000000000", "_cd": "1111:0000000", "_indextime": "5555555555", "_pre_msg": ( "04/23/2025 08:04:41 AM\nLogName=Test log\nSourceName=Server\nEventCode=0\nEventType=4\nType=Information\n" "ComputerName=#COMPUTERNAME#\nTaskCategory=Test log Server\nOpCode=Info\nRecordNumber=3\nKeywords=Classic" ), "_raw": ( "04/23/2025 08:04:41 AM\nLogName=Test log\nSourceName=Server\nEventCode=0\nEventType=4\nType=Information\n" "ComputerName=#COMPUTERNAME#\nTaskCategory=Test log Server\nOpCode=Info\nRecordNumber=3\nKeywords=Classic\n" "Message=Service started successfully.\n" ), "_serial": "1", "_si": ["ip-000-00-00-000", "main"], "_sourcetype": "WinEventLog", "_time": "2025-04-23T05:04:41.000-03:00", "host": "127.0.0.1", "index": "main", "linecount": "13", "source": "WinEventLog:Server", "sourcetype": "WinEventLog", "splunk_server": "ip-000-00-00-000", } ], [{"Indicator": "127.0.0.1", "Type": "hostname", "Vendor": "Splunk", "Score": 0, "isTypedIndicator": True}], ) mock_result_batch = io.BytesIO(bad_search_output) res = splunk.parse_batch_of_results(mock_result_batch, 10, "") assert res == expected_res @pytest.mark.parametrize( "args", [ {"entry_id": "entry_id"}, {"index": "invalid_index", "event": {"event_data": "event_data"}}, ], ) def test_splunk_submit_event_hec_command_invalid_index(mocker, requests_mock, args): """ Given: - An event to submit to Splunk via HEC with an invalid index. When: - Calling splunk_submit_event_hec_command. Then: - The function should not raise an exception - as we don't check for invalid indexes. """ from SplunkPyV2 import splunk_submit_event_hec_command from splunklib.client import Service mocker.patch("SplunkPyV2.get_events_from_file", return_value=[{"event": "test", "index": "invalid_index"}]) mocker.patch(RETURN_ERROR_TARGET) requests_mock.post("https://splunk.test.com/services/collector/event") service_mock = MagicMock(spec=Service) index_mock = MagicMock() index_mock.name = "valid_index" service_mock.indexes = [index_mock] splunk_submit_event_hec_command( params={"cred_hec_token": {"password": "token"}, "hec_url": "https://splunk.test.com"}, service=service_mock, args=args ) def test_get_modified_remote_data_skips_cached_events(mocker): """ Given: - An initial run of get_modified_remote_data_command processes an event. When: - get_modified_remote_data_command is called a second time, and the same event is mirrored again. Then: - Ensure the event from the second run is skipped because its key (event_id:timestamp) is already in the cache. """ from SplunkPyV2 import get_modified_remote_data_command test_id = "event_123" timestamp = "1737547610.49" event_key = f"{test_id}:{timestamp}" func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": False, "close_end_statuses": False, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(MagicMock(), False), } # Mock Splunk API responses audit_index_response = [ { "rule_id": test_id, "review_time": timestamp, "event_id": test_id, } ] # === First Run: Process and cache the event === mocker.patch("splunklib.results.JSONResultsReader", return_value=audit_index_response) mocker.patch("SplunkPyV2.get_integration_context", return_value={}) mocker.patch("SplunkPyV2.demisto.results") mocker.patch("SplunkPyV2.get_current_splunk_time", return_value="2021-02-09T17:41:30.589575+02:00") set_context_mock = mocker.patch("SplunkPyV2.set_integration_context") extensive_log_mock = mocker.patch("SplunkPyV2.extensive_log") get_modified_remote_data_command(MagicMock(), **func_call_kwargs) results = demisto.results.call_args[0][0][0]["Contents"] assert results["event_id"] == test_id # Assert the event was cached assert set_context_mock.call_count == 2 cached_context = set_context_mock.call_args[0][0] assert cached_context.get("processed_mirror_in_events_cache") == [event_key] # === Second Run: Should skip the cached event === mocker.patch("SplunkPyV2.get_integration_context", return_value=cached_context) set_context_mock.reset_mock() demisto.results.reset_mock() extensive_log_mock.reset_mock() get_modified_remote_data_command(MagicMock(), **func_call_kwargs) # Assert no new events were processed results = demisto.results.call_args[0][0] assert len(results) == 0 assert extensive_log_mock.call_args_list[0].contains("mirror-in: no findings was changed since") # ============================================================================================ # COMMENT HANDLING TESTS FOR get_modified_remote_data_command for supported Splunk ES versions # ============================================================================================ def test_get_modified_remote_data_with_multiple_notes(mocker): """ Test handling multiple Splunk note correctly. Given: - Splunk ES version is 8.0.0 or higher - Multiple notes exist for a finding in mc_notes When: - get_modified_remote_data_command is called Then: - All notes are fetched and processed - Multiple notes entries are created - notes are properly tagged and formatted """ from SplunkPyV2 import get_modified_remote_data_command test_id = "multi_notes_test" timestamp = "1737547610.49" func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": False, "close_end_statuses": False, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(MagicMock(), False), } # Mock incident_review response audit_index_response = [ { "rule_id": test_id, "review_time": timestamp, "event_id": test_id, } ] # Mock KV store response with multiple comments data mock_notes_data = [ { "notable_id": test_id, "note_id": "note_id_1", "content": "First note from mc_notes", "update_time": 1757409703.589575, "create_time": 1757409703, }, { "notable_id": test_id, "note_id": "note_id_2", "content": "Second note from mc_notes", "update_time": 1757409704.589575, "create_time": 1757409704, }, ] mocker.patch("SplunkPyV2.get_current_splunk_time", return_value="2021-02-09T17:41:30.589575+02:00") mocker.patch("SplunkPyV2.results.JSONResultsReader", side_effect=lambda res: res) # Mock KV store operations mock_service = MagicMock() mock_kv_store = MagicMock() mock_service.jobs.oneshot = ( lambda query, *args, **kwargs: audit_index_response if "source=notable_update_rest_handler" in query else mock_notes_data ) mock_kv_store.data.query.return_value = mock_notes_data mock_service.kvstore.__getitem__.return_value = mock_kv_store mocker.patch("SplunkPyV2.format_splunk_note_for_xsoar", side_effect=lambda note, _: note["content"]) mocker.patch("SplunkPyV2.get_integration_context", return_value={}) mocker.patch("SplunkPyV2.set_integration_context") results_mock = mocker.patch("SplunkPyV2.demisto.results") get_modified_remote_data_command(mock_service, **func_call_kwargs) # Verify results structure results = results_mock.call_args[0][0] assert len(results) == 3 # 1 finding entry + 2 comment entries # Verify finding entry finding_entry = results[0] assert finding_entry["EntryContext"]["mirrorRemoteId"] == test_id # Verify both note entries note_entries = results[1:] assert len(note_entries) == 2 for note_entry in note_entries: assert note_entry["Type"] == 1 assert note_entry["Tags"] == ["FROM SPLUNK"] assert note_entry["ContentsFormat"] == "markdown" assert "note from mc_notes" in note_entry["Contents"] def test_fetch_findings_with_notes(mocker): """ Test that fetch brings notes even when they're not part of the main fetch query results in ES 8.0+. Given: - Splunk ES version is 8.0.0 or higher - Fetch operation is running - Notes exist in mc_notes but not in the main finding search results When: - fetch_findings is called Then: - Notes are fetched via enrich_with_splunk_notes with is_fetch=True - Notes are stored in the finding under 'splunk_notes' key for incident creation - The fetch process includes notes even if they weren't in the original search """ from SplunkPyV2 import fetch_findings test_id = "fetch_comment_test" # Mock finding data without comments in the main search finding_data = { "rule_id": test_id, "event_id": test_id, "_time": "2021-02-09T16:41:30.589575+02:00", "rule_name": "Test Rule", "status": "new", } # Mock search results (no comments in main search) search_results = [finding_data] # Mock KV store operations for get_comments_data_new mock_notes_data = [ { "notable_id": test_id, "note_id": "note_id", "content": "notes from fetch", "update_time": 1757409703.589575, "create_time": 1757409703, } ] mock_kv_store = MagicMock() mock_kv_store.data.query.return_value = mock_notes_data # Mock helper functions. # `fetch_findings` now derives an epoch float from the fetch-window start # to scope the v2 KV-store note query (`enrich_with_splunk_notes_v2`), # so the helper must return real ISO timestamps (not empty strings). mocker.patch( "SplunkPyV2.get_fetch_time_window", return_value=( "2021-02-09T15:41:30.589575+02:00", "2021-02-09T17:41:30.589575+02:00", ), ) mocker.patch("SplunkPyV2.remove_irrelevant_incident_ids") mocker.patch("SplunkPyV2.format_splunk_note_for_xsoar", return_value="Note from fetch") # Mock other dependencies mocker.patch("SplunkPyV2.demisto.getLastRun", return_value={}) mocker.patch( "SplunkPyV2.demisto.params", return_value={ "fetchQuery": "search `notable`", "earliest_fetch_time_fieldname": "_time", "latest_fetch_time_fieldname": "_time", }, ) mocker.patch("SplunkPyV2.demisto.setLastRun") mocker.patch("SplunkPyV2.demisto.incidents") mocker.patch("SplunkPyV2.get_current_splunk_time", return_value="2021-02-09T17:41:30.589575+02:00") # Create mock service and mapper mock_service = MagicMock() mock_service.jobs.oneshot = lambda query, **kwargs: search_results if "search `notable`" in query else mock_notes_data mocker.patch("SplunkPyV2.results.JSONResultsReader", side_effect=lambda res: res) mock_service.kvstore.__getitem__.return_value = mock_kv_store mock_mapper = splunk.UserMappingObject(MagicMock(), False) # Call fetch_findings fetch_findings(service=mock_service, mapper=mock_mapper) # Verify that KV store was accessed (indicating get_comments_data_new was called) mock_service.kvstore.__getitem__.assert_called_with("mc_notes") # =========== update_remote_system_command Tests =========== @pytest.mark.parametrize( "delta, user_mapping_enabled, expected_owner, mapper_should_map", [ # Test case 1: Owner change with user mapping enabled and successful mapping ({"owner": "xsoar_user"}, True, "splunk_user", True), # Test case 2: Owner change with user mapping enabled but mapper returns None ({"owner": "xsoar_user"}, True, None, True), # Test case 3: Owner change with user mapping disabled ({"owner": "xsoar_user"}, False, None, False), # Test case 4: No owner change ({"status": "2"}, False, None, False), ], ) def test_update_remote_system_command_owner_mapping(mocker, delta, user_mapping_enabled, expected_owner, mapper_should_map): """ Test update_remote_system_command with different owner mapping scenarios. Given: - Different delta configurations with owner field - User mapping enabled/disabled - Mapper returning different values When: - update_remote_system_command is called Then: - Verify correct owner is passed to update_investigation_or_finding - Verify user mapping is called when enabled """ # Setup finding_id = "test_finding_123" args = { "remoteId": finding_id, "delta": delta, "data": {}, "entries": [], "incidentChanged": True, "status": 1, } params = {"userMapping": user_mapping_enabled, "close_finding": False} mock_service = MagicMock() mock_mapper = MagicMock() mock_mapper.should_map = mapper_should_map mock_mapper.get_splunk_user_by_xsoar.return_value = expected_owner mock_update = mocker.patch("SplunkPyV2.update_investigation_or_finding") mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "error") # Execute result = splunk.update_remote_system_command(args, params, mock_service, mock_mapper) # Verify assert result == finding_id if "owner" in delta and user_mapping_enabled and mapper_should_map: mock_mapper.get_splunk_user_by_xsoar.assert_called_once_with("xsoar_user") if expected_owner: mock_update.assert_called_once() call_kwargs = mock_update.call_args[1] assert call_kwargs["owner"] == expected_owner else: # When mapping returns None, error should be logged assert demisto.error.called @pytest.mark.parametrize( "delta, inc_status, close_finding_param, expected_status", [ # Test case 1: Incident closed and close_finding enabled ({"status": "2"}, 2, True, "5"), # IncidentStatus.DONE = 2 # Test case 2: Incident closed but close_finding disabled ({"status": "2"}, 2, False, "2"), # Test case 3: Incident not closed ({"status": "2"}, 1, True, "2"), # Test case 4: No status in delta but incident closed ({"urgency": "high"}, 2, True, "5"), ], ) def test_update_remote_system_command_close_finding(mocker, delta, inc_status, close_finding_param, expected_status): """ Test update_remote_system_command closing finding functionality. Given: - Different incident statuses - close_finding parameter enabled/disabled - Different delta configurations When: - update_remote_system_command is called Then: - Verify status is set to "5" (closed) when appropriate """ # Setup finding_id = "test_finding_456" args = { "remoteId": finding_id, "delta": delta, "data": {}, "entries": [], "incidentChanged": True, "status": inc_status, } params = {"userMapping": False, "close_finding": close_finding_param} mock_service = MagicMock() mock_mapper = MagicMock() mock_mapper.should_map = False mock_update = mocker.patch("SplunkPyV2.update_investigation_or_finding") mocker.patch.object(demisto, "debug") # Execute result = splunk.update_remote_system_command(args, params, mock_service, mock_mapper) # Verify assert result == finding_id if mock_update.called: call_kwargs = mock_update.call_args[1] assert call_kwargs.get("status") == expected_status def test_update_remote_system_command_multiple_fields(mocker): """ Test update_remote_system_command with multiple field updates. Given: - Delta with multiple mirrored fields (status, urgency, disposition) When: - update_remote_system_command is called Then: - All fields are passed to update_investigation_or_finding """ # Setup finding_id = "test_finding_789" delta = { "status": "2", "urgency": "high", "disposition": "disposition:1", "reviewer": "test_reviewer", } args = { "remoteId": finding_id, "delta": delta, "data": {}, "entries": [], "incidentChanged": True, "status": 1, } params = {"userMapping": False, "close_finding": False} mock_service = MagicMock() mock_mapper = MagicMock() mock_mapper.should_map = False mock_update = mocker.patch("SplunkPyV2.update_investigation_or_finding") mocker.patch.object(demisto, "debug") # Execute result = splunk.update_remote_system_command(args, params, mock_service, mock_mapper) # Verify assert result == finding_id mock_update.assert_called_once() call_kwargs = mock_update.call_args[1] assert call_kwargs["status"] == "2" assert call_kwargs["urgency"] == "high" assert call_kwargs["disposition"] == "disposition:1" def test_update_remote_system_command_with_note_in_delta(mocker): """ Test update_remote_system_command with note in delta. Given: - Delta containing a note field When: - update_remote_system_command is called Then: - add_investigation_note is called with the note content - Note includes COMMENT_MIRRORED_FROM_XSOAR marker """ # Setup finding_id = "test_finding_note" note_content = "This is a test note" delta = {"note": note_content} args = { "remoteId": finding_id, "delta": delta, "data": {}, "entries": [], "incidentChanged": True, "status": 1, } params = {"userMapping": False, "close_finding": False} mock_service = MagicMock() mock_mapper = MagicMock() mock_mapper.should_map = False mocker.patch("SplunkPyV2.update_investigation_or_finding") mock_add_note = mocker.patch("SplunkPyV2.add_investigation_note") mocker.patch.object(demisto, "debug") # Execute result = splunk.update_remote_system_command(args, params, mock_service, mock_mapper) # Verify assert result == finding_id mock_add_note.assert_called_once() call_args = mock_add_note.call_args assert call_args[1]["investigation_or_finding_id"] == finding_id assert note_content in call_args[1]["content"] assert splunk.COMMENT_MIRRORED_FROM_XSOAR in call_args[1]["content"] def test_update_remote_system_command_with_entries(mocker): """ Test update_remote_system_command with entries containing notes. Given: - Entries with NOTE_TAG_TO_SPLUNK tag When: - update_remote_system_command is called Then: - add_investigation_note is called for each tagged entry """ # Setup finding_id = "test_finding_entries" entry1_content = "Entry 1 content" entry2_content = "Entry 2 content" entries = [ {"tags": [splunk.NOTE_TAG_TO_SPLUNK], "contents": entry1_content}, {"tags": ["OTHER_TAG"], "contents": "Should not be added"}, {"tags": [splunk.NOTE_TAG_TO_SPLUNK], "contents": entry2_content}, ] args = { "remoteId": finding_id, "delta": {}, "data": {}, "entries": entries, "incidentChanged": False, "status": 1, } params = {"userMapping": False, "close_finding": False} mock_service = MagicMock() mock_mapper = MagicMock() mock_add_note = mocker.patch("SplunkPyV2.add_investigation_note") mocker.patch.object(demisto, "debug") # Execute result = splunk.update_remote_system_command(args, params, mock_service, mock_mapper) # Verify assert result == finding_id assert mock_add_note.call_count == 2 # Verify first call first_call = mock_add_note.call_args_list[0] assert entry1_content in first_call[1]["content"] assert splunk.COMMENT_MIRRORED_FROM_XSOAR in first_call[1]["content"] # Verify second call second_call = mock_add_note.call_args_list[1] assert entry2_content in second_call[1]["content"] assert splunk.COMMENT_MIRRORED_FROM_XSOAR in second_call[1]["content"] def test_update_remote_system_command_no_changes(mocker): """ Test update_remote_system_command when incident hasn't changed. Given: - incidentChanged is False When: - update_remote_system_command is called Then: - No API calls are made - Finding ID is still returned """ # Setup finding_id = "test_finding_no_change" args = { "remoteId": finding_id, "delta": {"status": "2"}, "data": {}, "entries": [], "incidentChanged": False, "status": 1, } params = {"userMapping": False, "close_finding": False} mock_service = MagicMock() mock_mapper = MagicMock() mock_update = mocker.patch("SplunkPyV2.update_investigation_or_finding") mocker.patch.object(demisto, "debug") # Execute result = splunk.update_remote_system_command(args, params, mock_service, mock_mapper) # Verify assert result == finding_id mock_update.assert_not_called() def test_update_remote_system_command_empty_delta(mocker): """ Test update_remote_system_command with empty delta. Given: - Empty delta dictionary - incidentChanged is True When: - update_remote_system_command is called Then: - No API calls are made (no changed data) """ # Setup finding_id = "test_finding_empty_delta" args = { "remoteId": finding_id, "delta": {}, "data": {}, "entries": [], "incidentChanged": True, "status": 1, } params = {"userMapping": False, "close_finding": False} mock_service = MagicMock() mock_mapper = MagicMock() mock_update = mocker.patch("SplunkPyV2.update_investigation_or_finding") mocker.patch.object(demisto, "debug") # Execute result = splunk.update_remote_system_command(args, params, mock_service, mock_mapper) # Verify assert result == finding_id mock_update.assert_not_called() def test_update_remote_system_command_api_error(mocker): """ Test update_remote_system_command when API call fails. Given: - Valid delta with changes - update_investigation_or_finding raises an exception When: - update_remote_system_command is called Then: - Error is logged - Finding ID is still returned """ # Setup finding_id = "test_finding_error" delta = {"status": "2"} args = { "remoteId": finding_id, "delta": delta, "data": {}, "entries": [], "incidentChanged": True, "status": 1, } params = {"userMapping": False, "close_finding": False} mock_service = MagicMock() mock_mapper = MagicMock() mock_mapper.should_map = False error_message = "API Error: Connection failed" mock_update = mocker.patch("SplunkPyV2.update_investigation_or_finding", side_effect=Exception(error_message)) mock_error = mocker.patch.object(demisto, "error") mocker.patch.object(demisto, "debug") # Execute result = splunk.update_remote_system_command(args, params, mock_service, mock_mapper) # Verify assert result == finding_id mock_update.assert_called_once() mock_error.assert_called() error_call_args = mock_error.call_args[0][0] assert finding_id in error_call_args assert error_message in error_call_args def test_update_remote_system_command_note_api_error(mocker): """ Test update_remote_system_command when add_investigation_note fails. Given: - Delta with note field - add_investigation_note raises an exception When: - update_remote_system_command is called Then: - Error is logged - Finding ID is still returned """ # Setup finding_id = "test_finding_note_error" delta = {"note": "Test note"} args = { "remoteId": finding_id, "delta": delta, "data": {}, "entries": [], "incidentChanged": True, "status": 1, } params = {"userMapping": False, "close_finding": False} mock_service = MagicMock() mock_mapper = MagicMock() mock_mapper.should_map = False mocker.patch("SplunkPyV2.update_investigation_or_finding") error_message = "Note API Error" mock_add_note = mocker.patch("SplunkPyV2.add_investigation_note", side_effect=Exception(error_message)) mock_error = mocker.patch.object(demisto, "error") mocker.patch.object(demisto, "debug") # Execute result = splunk.update_remote_system_command(args, params, mock_service, mock_mapper) # Verify assert result == finding_id mock_add_note.assert_called_once() mock_error.assert_called() error_call_args = mock_error.call_args[0][0] assert finding_id in error_call_args def test_update_remote_system_command_non_mirrored_fields_ignored(mocker): """ Test that non-mirrored fields in delta are ignored. Given: - Delta with both mirrored and non-mirrored fields When: - update_remote_system_command is called Then: - Only mirrored fields are passed to the API """ # Setup finding_id = "test_finding_filtered" delta = { "status": "2", # Mirrored "urgency": "high", # Mirrored "custom_field": "value", # Not mirrored "another_field": "test", # Not mirrored } args = { "remoteId": finding_id, "delta": delta, "data": {}, "entries": [], "incidentChanged": True, "status": 1, } params = {"userMapping": False, "close_finding": False} mock_service = MagicMock() mock_mapper = MagicMock() mock_mapper.should_map = False mock_update = mocker.patch("SplunkPyV2.update_investigation_or_finding") mocker.patch.object(demisto, "debug") # Execute result = splunk.update_remote_system_command(args, params, mock_service, mock_mapper) # Verify assert result == finding_id mock_update.assert_called_once() call_kwargs = mock_update.call_args[1] # Verify only mirrored fields are present assert call_kwargs["status"] == "2" assert call_kwargs["urgency"] == "high" assert "custom_field" not in str(call_kwargs) assert "another_field" not in str(call_kwargs) @pytest.mark.parametrize( "splunk_time, expected_offset", [ # Valid positive timezone offsets ("2024-01-15T10:30:45.123456+02:00", "+02:00"), ("2024-01-15T10:30:45.123456+09:30", "+09:30"), ("2024-01-15T10:30:45.123456+14:00", "+14:00"), # Valid negative timezone offsets ("2024-01-15T10:30:45.123456-05:00", "-05:00"), ("2024-01-15T10:30:45.123456-03:30", "-03:30"), ("2024-01-15T10:30:45.123456-12:00", "-12:00"), # UTC/zero offset ("2024-01-15T10:30:45.123456+00:00", "+00:00"), # Different microsecond precision ("2024-01-15T10:30:45.1+02:00", "+02:00"), ("2024-01-15T10:30:45.12+02:00", "+02:00"), ("2024-01-15T10:30:45.123+02:00", "+02:00"), ("2024-01-15T10:30:45.1234+02:00", "+02:00"), ("2024-01-15T10:30:45.12345+02:00", "+02:00"), ], ) def test_extract_timezone_offset_from_splunk_time_valid_inputs(splunk_time, expected_offset): """ Given: A valid Splunk time string in ISO_FORMAT_TZ_AWARE format with various timezone offsets. When: The extract_timezone_offset_from_splunk_time function is called. Then: The function returns the correct timezone offset string. """ result = splunk.extract_timezone_offset_from_splunk_time(splunk_time) assert result == expected_offset @pytest.mark.parametrize( "splunk_time, expected_offset", [ # Empty string ("", "+00:00"), # Malformed strings ("invalid-time-string", "+00:00"), ("2024-01-15", "+00:00"), ("2024-01-15T10:30:45", "+00:00"), # Missing timezone ("2024-01-15T10:30:45.123456", "+00:00"), # Z notation (not in expected format) ("2024-01-15T10:30:45.123456Z", "+00:00"), # Invalid timezone format ("2024-01-15T10:30:45.123456+2:00", "+00:00"), ("2024-01-15T10:30:45.123456+0200", "+02:00"), ], ) def test_extract_timezone_offset_from_splunk_time_invalid_inputs(mocker, splunk_time, expected_offset): """ Given: An invalid or malformed Splunk time string. When: The extract_timezone_offset_from_splunk_time function is called. Then: The function returns '+00:00' (UTC) as the default fallback. """ mocker.patch.object(demisto, "error") result = splunk.extract_timezone_offset_from_splunk_time(splunk_time) assert result == expected_offset def test_extract_timezone_offset_from_splunk_time_edge_cases(): """ Given: Edge case Splunk time strings with boundary timezone values. When: The extract_timezone_offset_from_splunk_time function is called. Then: The function correctly handles extreme timezone offsets. """ # Maximum positive offset (UTC+14:00) result = splunk.extract_timezone_offset_from_splunk_time("2024-01-15T10:30:45.123456+14:00") assert result == "+14:00" # Maximum negative offset (UTC-12:00) result = splunk.extract_timezone_offset_from_splunk_time("2024-01-15T10:30:45.123456-12:00") assert result == "-12:00" # Half-hour offset result = splunk.extract_timezone_offset_from_splunk_time("2024-01-15T10:30:45.123456+05:30") assert result == "+05:30" # Quarter-hour offset result = splunk.extract_timezone_offset_from_splunk_time("2024-01-15T10:30:45.123456+05:45") assert result == "+05:45" def test_splunk_get_indexes_command_success(mocker): """ Given: - A Splunk service object. - The REST API query for indexes succeeds. When: - calling splunk_get_indexes_command. Then: - Ensure the command returns the expected indexes from the REST API query. - Ensure the fallback mechanism (service.indexes) is NOT used. """ from SplunkPyV2 import splunk_get_indexes_command # Mock the service and the oneshot job results service = mocker.MagicMock() mock_result = {"name": "main", "count": "100"} mocker.patch("splunklib.results.JSONResultsReader", return_value=[mock_result]) # Mock return_results to capture the output return_results_mock = mocker.patch("SplunkPyV2.return_results") # Call the function splunk_get_indexes_command(service, "search") # Verify results assert return_results_mock.call_count == 1 results = return_results_mock.call_args[0][0] assert results.raw_response == json.dumps([mock_result]) # Verify oneshot was called service.jobs.oneshot.assert_called_once() def test_splunk_get_indexes_command_fallback(mocker): """ Given: - A Splunk service object. - The REST API query for indexes fails. - The direct API (service.indexes) succeeds. When: - calling splunk_get_indexes_command. Then: - Ensure the command returns the expected indexes from the direct API. - Ensure the error is logged and fallback is attempted. """ from SplunkPyV2 import splunk_get_indexes_command # Mock the service service = mocker.MagicMock() # Mock oneshot to raise an exception service.jobs.oneshot.side_effect = Exception("REST API Failed") # Mock service.indexes to return a list of indexes mock_index = mocker.MagicMock() mock_index.name = "history" # Mocking dictionary access for the index object since the code uses index["totalEventCount"] mock_index.__getitem__.return_value = "50" service.indexes = [mock_index] # Mock logging and return_results error_mock = mocker.patch("demistomock.error") debug_mock = mocker.patch("demistomock.debug") return_results_mock = mocker.patch("SplunkPyV2.return_results") # Call the function splunk_get_indexes_command(service, "search") # Verify error was logged assert error_mock.call_count == 1 assert "Failed to get indexes using REST API query approach" in error_mock.call_args[0][0] # Verify fallback was attempted (debug log) fallback_log_found = False for call in debug_mock.call_args_list: if "Falling back to direct API approach" in call[0][0]: fallback_log_found = True break assert fallback_log_found # Verify results expected_result = [{"name": "history", "count": "50"}] assert return_results_mock.call_count == 1 results = return_results_mock.call_args[0][0] assert results.raw_response == json.dumps(expected_result) def test_splunk_get_indexes_command_failure(mocker): """ Given: - A Splunk service object. - Both the REST API query and the direct API fail. When: - calling splunk_get_indexes_command. Then: - Ensure a DemistoException is raised with details from both errors. """ from SplunkPyV2 import splunk_get_indexes_command # Mock the service service = mocker.MagicMock() # Mock oneshot to raise an exception service.jobs.oneshot.side_effect = Exception("REST API Failed") # Mock service.indexes to raise an exception (property access raises exception) type(service).indexes = mocker.PropertyMock(side_effect=Exception("Direct API Failed")) # Mock logging error_mock = mocker.patch("demistomock.error") # Call the function and expect DemistoException with pytest.raises(DemistoException) as e: splunk_get_indexes_command(service, "search") assert "Failed to retrieve indexes using both methods" in str(e.value) assert "REST API error: REST API Failed" in str(e.value) assert "Direct API error: Direct API Failed" in str(e.value) # Verify errors were logged assert error_mock.call_count >= 2 # One for REST failure, one for Direct failure # ========== unique_id_fields Tests ========== @pytest.mark.parametrize( "unique_id_fields, incident_data, expected_fields_in_id", [ # Test 1: No unique_id_fields parameter (default behavior) ( "", { "rawJSON": json.dumps( { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "test raw data", } ) }, ["_cd", "index", "_time", "_indextime", "_raw"], ), # Test 2: With single unique_id_field ( "source", { "rawJSON": json.dumps( { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "test raw data", "source": "test_source", } ) }, ["_cd", "index", "_time", "_indextime", "_raw", "source"], ), # Test 3: With multiple unique_id_fields ( "source,host,event_type", { "rawJSON": json.dumps( { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "test raw data", "source": "test_source", "host": "test_host", "event_type": "test_type", } ) }, ["_cd", "index", "_time", "_indextime", "_raw", "source", "host", "event_type"], ), # Test 4: With unique_id_fields that don't exist in data (should not break) ( "nonexistent_field", { "rawJSON": json.dumps( { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "test raw data", } ) }, ["_cd", "index", "_time", "_indextime", "_raw"], ), # Test 5: With unique_id_fields containing spaces (should be trimmed) ( " source , host ", { "rawJSON": json.dumps( { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "test raw data", "source": "test_source", "host": "test_host", } ) }, ["_cd", "index", "_time", "_indextime", "_raw", " source ", " host "], ), ], ids=[ "no_unique_id_fields", "single_unique_id_field", "multiple_unique_id_fields", "nonexistent_unique_id_field", "unique_id_fields_with_spaces", ], ) def test_create_incident_custom_id_with_unique_fields(mocker, unique_id_fields, incident_data, expected_fields_in_id): """ Test the create_incident_custom_id function with various unique_id_fields configurations. Given: - Different configurations of the unique_id_fields parameter - Incident data with various fields When: - create_incident_custom_id is called Then: - Verify that the custom ID is generated correctly - Verify that all expected fields are included in the ID generation - Verify that the function handles missing fields gracefully """ mocker.patch.object(demisto, "params", return_value={"unique_id_fields": unique_id_fields}) mocker.patch.object(demisto, "debug") # Call the function custom_id = splunk.create_incident_custom_id(incident_data) # Verify the ID is a valid MD5 hash assert len(custom_id) == 32 assert all(c in "0123456789abcdef" for c in custom_id) # Verify that the function was called with params demisto.params.assert_called() def test_create_incident_custom_id_generates_unique_ids(mocker): """ Test that create_incident_custom_id generates different IDs for different incidents. Given: - Two incidents with different data - unique_id_fields parameter configured When: - create_incident_custom_id is called for both incidents Then: - Verify that different IDs are generated for different incidents - Verify that the same incident generates the same ID consistently """ mocker.patch.object(demisto, "params", return_value={"unique_id_fields": "source,host"}) mocker.patch.object(demisto, "debug") incident1 = { "rawJSON": json.dumps( { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "test raw data 1", "source": "source1", "host": "host1", } ) } incident2 = { "rawJSON": json.dumps( { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "test raw data 2", "source": "source2", "host": "host2", } ) } # Generate IDs id1_first = splunk.create_incident_custom_id(incident1) id2 = splunk.create_incident_custom_id(incident2) id1_second = splunk.create_incident_custom_id(incident1) # Different incidents should have different IDs assert id1_first != id2 # Same incident should generate the same ID assert id1_first == id1_second def test_create_incident_custom_id_prevents_duplicates(mocker): """ Test that unique_id_fields helps prevent duplicate IDs. Given: - Two incidents with same default fields but different unique_id_fields values - unique_id_fields parameter configured When: - create_incident_custom_id is called for both incidents Then: - Verify that different IDs are generated when unique_id_fields differ """ mocker.patch.object(demisto, "params", return_value={"unique_id_fields": "source"}) mocker.patch.object(demisto, "debug") # Same default fields, different source incident1 = { "rawJSON": json.dumps( { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "same raw data", "source": "source_A", } ) } incident2 = { "rawJSON": json.dumps( { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "same raw data", "source": "source_B", } ) } id1 = splunk.create_incident_custom_id(incident1) id2 = splunk.create_incident_custom_id(incident2) # Should generate different IDs due to different source values assert id1 != id2 @pytest.mark.parametrize( "unique_id_fields", [ None, "", " ", ], ids=["None", "empty_string", "whitespace"], ) def test_create_incident_custom_id_with_empty_unique_fields(mocker, unique_id_fields): """ Test that create_incident_custom_id handles empty/None unique_id_fields gracefully. Given: - unique_id_fields parameter is None, empty string, or whitespace When: - create_incident_custom_id is called Then: - Verify that the function uses only default fields - Verify that a valid ID is still generated """ mocker.patch.object(demisto, "params", return_value={"unique_id_fields": unique_id_fields}) mocker.patch.object(demisto, "debug") incident = { "rawJSON": json.dumps( { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "test raw data", } ) } custom_id = splunk.create_incident_custom_id(incident) # Should still generate a valid MD5 hash assert len(custom_id) == 32 assert all(c in "0123456789abcdef" for c in custom_id) def test_test_module_duplicate_detection_with_unique_fields(mocker): """ Test that test_module properly detects duplicates and suggests using unique_id_fields. Given: - Fetch query that returns duplicate incident IDs - No unique_id_fields configured When: - test_module is executed Then: - Verify that an error is raised - Verify that the error message mentions the unique_id_fields parameter """ mocker.patch.object( demisto, "params", return_value={"isFetch": True, "fetchQuery": "search test | table index", "unique_id_fields": ""} ) # Mock service and results with duplicate IDs service = mocker.MagicMock() # Create incidents that will generate duplicate IDs duplicate_incidents = [ { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "same data", "unique_field": "event1", }, { "_cd": "668:17198", "index": "finding", "_time": "2020-08-04T05:45:16.000-07:00", "_indextime": "1596545116", "_raw": "same data", "unique_field": "event2", }, ] mocker.patch("splunklib.results.JSONResultsReader", return_value=duplicate_incidents) return_error_mock = mocker.patch(RETURN_ERROR_TARGET) # Execute test_module splunk.test_module(service, demisto.params()) # Verify error was raised assert return_error_mock.called error_message = return_error_mock.call_args[0][0] # Verify error message mentions unique_id_fields assert "Unique ID Fields" in error_message or "unique_id_fields" in error_message assert "integration configuration" in error_message.lower() or "integration settings" in error_message.lower() # ========== ResponseSizeValidator Tests ========== def test_response_size_validator_no_warning_below_threshold(mocker): """ Given: - Data size below the 20 MB threshold When: - ResponseSizeValidator validates the data Then: - return_results is not called - validated flag is set to True """ validator = splunk.ResponseSizeValidator() mock_return_results = mocker.patch("SplunkPyV2.return_results") # Create data below threshold (1 MB) - as list of dicts small_data = [{"data": "x" * (1 * 1024 * 1024)}] validator.validate_and_report(small_data) mock_return_results.assert_not_called() assert validator.validated is True def test_response_size_validator_warning_above_threshold(mocker): """ Given: - Data size above the 20 MB threshold (e.g., 25 MB) When: - ResponseSizeValidator validates the data Then: - return_results is called with warning message - validated flag is set to True - Warning message contains "WARNING" prefix """ validator = splunk.ResponseSizeValidator() mock_return_results = mocker.patch("SplunkPyV2.return_results") # Create data above threshold (25 MB) - as list of dicts large_data = [{"data": "x" * (25 * 1024 * 1024)}] validator.validate_and_report(large_data) mock_return_results.assert_called_once() warning = mock_return_results.call_args[0][0] assert "WARNING" in warning assert "25." in warning # Size will be around 25 MB assert "20" in warning assert "normal usage size" in warning assert validator.validated is True def test_response_size_validator_warning_shown_only_once(mocker): """ Given: - Multiple batches of data, all above threshold When: - ResponseSizeValidator validates each batch Then: - return_results is called only on the first validation - Subsequent validations don't call return_results - validated flag remains True after first validation """ validator = splunk.ResponseSizeValidator() mock_return_results = mocker.patch("SplunkPyV2.return_results") # Create data above threshold (25 MB) - as list of dicts large_data = [{"data": "x" * (25 * 1024 * 1024)}] # First validation should call return_results validator.validate_and_report(large_data) assert mock_return_results.call_count == 1 assert validator.validated is True # Second validation should not call return_results (already validated) validator.validate_and_report(large_data) assert mock_return_results.call_count == 1 assert validator.validated is True # Third validation should also not call return_results validator.validate_and_report(large_data) assert mock_return_results.call_count == 1 assert validator.validated is True def test_response_size_validator_just_above_threshold(mocker): """ Given: - Data size just above the 20 MB threshold (20 MB + 1 byte) When: - ResponseSizeValidator validates the data Then: - return_results is called with warning message - validated flag is set to True """ validator = splunk.ResponseSizeValidator() mock_return_results = mocker.patch("SplunkPyV2.return_results") # Create data just above threshold (20 MB + 1 byte) - as list of dicts just_above_data = [{"data": "x" * (20 * 1024 * 1024 + 1)}] validator.validate_and_report(just_above_data) mock_return_results.assert_called_once() warning = mock_return_results.call_args[0][0] assert "WARNING" in warning assert validator.validated is True def test_response_size_validator_message_format(mocker): """ Given: - Data size of 30 MB (above threshold) When: - ResponseSizeValidator validates the data Then: - Warning message contains all required elements: * "WARNING" prefix * Actual size in MB * Threshold size in MB * "normal usage size" phrase """ validator = splunk.ResponseSizeValidator() mock_return_results = mocker.patch("SplunkPyV2.return_results") # Create data of 30 MB - as list of dicts data_30mb = [{"data": "x" * (30 * 1024 * 1024)}] validator.validate_and_report(data_30mb) mock_return_results.assert_called_once() warning = mock_return_results.call_args[0][0] assert warning.startswith("WARNING:") assert "30." in warning # Size will be around 30 MB assert "20" in warning assert "normal usage size" in warning assert "exceeds" in warning.lower() # ===================== Tests for add_investigation_note retry/fallback ===================== def test_add_investigation_note_success_without_notable_time(): """ Given: - A mock Splunk service and valid note parameters. When: - add_investigation_note is called and the first service.post call succeeds. Then: - The result matches the expected response. - service.post is called exactly once (without notable_time). """ mock_service = MagicMock() expected_result = {"id": "note-123", "content": "test note"} mock_response = MagicMock() mock_response.body.read.return_value = json.dumps(expected_result).encode() mock_service.post.return_value = mock_response result = splunk.add_investigation_note( service=mock_service, investigation_or_finding_id="finding-abc", content="test note", ) assert result == expected_result mock_service.post.assert_called_once_with( "public/v2/investigations/finding-abc/notes", body=json.dumps({"content": "test note"}), ) def test_add_investigation_note_fallback_with_notable_time(): """ Given: - A mock Splunk service where the first service.post call raises an exception. When: - add_investigation_note is called. Then: - The function retries with notable_time=now and returns the result from the second call. - service.post is called exactly twice. - The second call includes notable_time="now". """ mock_service = MagicMock() expected_result = {"id": "note-456", "content": "fallback note"} mock_response = MagicMock() mock_response.body.read.return_value = json.dumps(expected_result).encode() mock_service.post.side_effect = [Exception("API error"), mock_response] result = splunk.add_investigation_note( service=mock_service, investigation_or_finding_id="finding-abc", content="fallback note", ) assert result == expected_result assert mock_service.post.call_count == 2 second_call = mock_service.post.call_args_list[1] assert second_call.kwargs.get("notable_time") == "now" def test_add_investigation_note_both_fail(): """ Given: - A mock Splunk service where both service.post calls raise exceptions. When: - add_investigation_note is called. Then: - The second exception propagates to the caller. - service.post is called exactly twice. """ mock_service = MagicMock() mock_service.post.side_effect = [Exception("First error"), Exception("Second error")] with pytest.raises(Exception, match="Second error"): splunk.add_investigation_note( service=mock_service, investigation_or_finding_id="finding-abc", content="will fail", ) assert mock_service.post.call_count == 2 def test_add_investigation_note_with_note_type(): """ Given: - A mock Splunk service and a note_type parameter is provided. When: - add_investigation_note is called with note_type="Task". Then: - The request body includes both "content" and "type" fields. - service.post is called once with the correct body. """ mock_service = MagicMock() expected_result = {"id": "note-789", "content": "typed note", "type": "Task"} mock_response = MagicMock() mock_response.body.read.return_value = json.dumps(expected_result).encode() mock_service.post.return_value = mock_response result = splunk.add_investigation_note( service=mock_service, investigation_or_finding_id="finding-xyz", content="typed note", note_type="Task", ) assert result == expected_result mock_service.post.assert_called_once_with( "public/v2/investigations/finding-xyz/notes", body=json.dumps({"content": "typed note", "type": "Task"}), ) # ===================== Tests for update_investigation_or_finding retry/fallback ===================== def test_update_investigation_or_finding_success_without_notable_time(): """ Given: - A mock Splunk service and valid update fields. When: - update_investigation_or_finding is called and the first service.post call succeeds. Then: - The result matches the expected response. - service.post is called exactly once (without notable_time). """ mock_service = MagicMock() expected_result = {"id": "finding-abc", "status": "closed"} mock_response = MagicMock() mock_response.body.read.return_value = json.dumps(expected_result).encode() mock_service.post.return_value = mock_response result = splunk.update_investigation_or_finding( service=mock_service, investigation_or_finding_id="finding-abc", status="closed", ) assert result == expected_result mock_service.post.assert_called_once_with( "public/v2/investigations/finding-abc", body=json.dumps({"status": "closed"}), ) def test_update_investigation_or_finding_fallback_with_notable_time(): """ Given: - A mock Splunk service where the first service.post call raises an exception. When: - update_investigation_or_finding is called. Then: - The function retries with notable_time=now and returns the result from the second call. - service.post is called exactly twice. - The second call includes notable_time="now". """ mock_service = MagicMock() expected_result = {"id": "finding-abc", "owner": "admin"} mock_response = MagicMock() mock_response.body.read.return_value = json.dumps(expected_result).encode() mock_service.post.side_effect = [Exception("API error"), mock_response] result = splunk.update_investigation_or_finding( service=mock_service, investigation_or_finding_id="finding-abc", owner="admin", ) assert result == expected_result assert mock_service.post.call_count == 2 second_call = mock_service.post.call_args_list[1] assert second_call.kwargs.get("notable_time") == "now" def test_update_investigation_or_finding_both_fail(mocker): """ Given: - A mock Splunk service where both service.post calls raise exceptions. When: - update_investigation_or_finding is called. Then: - The second Exception is raised (from the retry attempt). - service.post is called exactly twice. """ mock_service = MagicMock() mock_service.post.side_effect = [Exception("First error"), Exception("Second error")] with pytest.raises(Exception, match="Second error"): splunk.update_investigation_or_finding( service=mock_service, investigation_or_finding_id="finding-abc", status="closed", ) assert mock_service.post.call_count == 2 def test_update_investigation_or_finding_no_fields(): """ Given: - A mock Splunk service and no update fields provided (all None). When: - update_investigation_or_finding is called without any fields. Then: - Returns a dict with success=False and a message about no fields. - service.post is never called. """ mock_service = MagicMock() result = splunk.update_investigation_or_finding( service=mock_service, investigation_or_finding_id="finding-abc", ) assert result == {"success": False, "message": "No fields provided to update"} mock_service.post.assert_not_called() def test_update_investigation_or_finding_partial_fields(): """ Given: - A mock Splunk service and only some update fields provided. When: - update_investigation_or_finding is called with owner and urgency (but not status or disposition). Then: - The request body contains only the provided fields. - service.post is called once with the correct partial body. """ mock_service = MagicMock() expected_result = {"id": "finding-abc", "owner": "admin", "urgency": "high"} mock_response = MagicMock() mock_response.body.read.return_value = json.dumps(expected_result).encode() mock_service.post.return_value = mock_response result = splunk.update_investigation_or_finding( service=mock_service, investigation_or_finding_id="finding-abc", owner="admin", urgency="high", ) assert result == expected_result mock_service.post.assert_called_once_with( "public/v2/investigations/finding-abc", body=json.dumps({"owner": "admin", "urgency": "high"}), ) def test_update_investigation_or_finding_with_finding_time(): """ Given: - A mock Splunk service, valid update fields, and a finding_time value. When: - update_investigation_or_finding is called with finding_time provided and the first service.post call succeeds. Then: - service.post is called exactly once. - The call includes notable_time equal to the provided finding_time (no fallback to "now"). - The result matches the expected response. """ mock_service = MagicMock() expected_result = {"id": "finding-abc", "status": "closed"} mock_response = MagicMock() mock_response.body.read.return_value = json.dumps(expected_result).encode() mock_service.post.return_value = mock_response finding_time = "2024-01-15T12:34:56.000+00:00" result = splunk.update_investigation_or_finding( service=mock_service, investigation_or_finding_id="finding-abc", status="closed", finding_time=finding_time, ) assert result == expected_result mock_service.post.assert_called_once_with( "public/v2/investigations/finding-abc", body=json.dumps({"status": "closed"}), notable_time=finding_time, ) # ================================================================================= # Phase 3a — Fetch Investigations scaffolding (helpers + model + factory). # These tests cover only the new (currently dead-code) pieces added in Phase 3a. # Existing fetch behavior is unchanged. # ================================================================================= # --------------------------- prepare_investigations_query --------------------------- class TestPrepareInvestigationsQuery: """Given the user's `investigations_fetch_query` SPL, when prepare_investigations_query runs, then the mandatory `FETCH_FILTER_PLACEHOLDER` token is substituted with the four managed params; if the token is missing, a DemistoException is raised. """ DEFAULT_URL = "/servicesNS/nobody/missioncontrol/public/v2/investigations?search_format=true" "&FETCH_FILTER_PLACEHOLDER" DEFAULT_QUERY = f'| rest "{DEFAULT_URL}"' def test_given_placeholder_when_called_then_substituted_with_all_four_params(self): """Given the default SPL containing FETCH_FILTER_PLACEHOLDER, when prepare_investigations_query runs, then the placeholder is replaced and all four managed params appear once (timestamps are URL-encoded by ``urlencode``, so ':' becomes '%3A').""" out = splunk.prepare_investigations_query( user_query=self.DEFAULT_QUERY, create_time_min="2025-01-01T00:00:00Z", create_time_max="2025-01-02T00:00:00Z", limit=50, offset=0, ) assert "FETCH_FILTER_PLACEHOLDER" not in out for fragment in ( "create_time_min=2025-01-01T00%3A00%3A00Z", "create_time_max=2025-01-02T00%3A00%3A00Z", "limit=50", "offset=0", ): assert out.count(fragment) == 1, f"expected exactly one '{fragment}' in {out!r}" def test_given_query_without_placeholder_when_called_then_raises_demisto_exception(self): """Given a user_query that does NOT contain FETCH_FILTER_PLACEHOLDER, when prepare_investigations_query runs, then a DemistoException is raised whose message names the missing token and the parameter to fix.""" query = '| rest "/path/v2/investigations?search_format=true"' with pytest.raises(splunk.DemistoException) as exc_info: splunk.prepare_investigations_query(query, "2025-01-01T00:00:00Z", "2025-01-02T00:00:00Z", 10, 0) msg = str(exc_info.value) assert "FETCH_FILTER_PLACEHOLDER" in msg assert "Investigations fetch query" in msg def test_given_limit_above_cap_when_called_then_clamped_to_100(self): """Given `limit > 100`, when called, then it is clamped to INVESTIGATIONS_MAX_LIMIT (100).""" out = splunk.prepare_investigations_query( self.DEFAULT_QUERY, "2025-01-01T00:00:00Z", "2025-01-02T00:00:00Z", limit=999, offset=0 ) assert "limit=100" in out assert "limit=999" not in out @pytest.mark.parametrize( "ts_min, ts_max", [ ("2025-01-01T00:00:00Z", "2025-01-02T00:00:00Z"), ("2026-04-29T10:00:00.3950565Z", "2026-04-30T10:00:00.3950565Z"), ], ) def test_given_iso_boundary_strings_when_substituted_then_passed_through(self, ts_min, ts_max): """Given seconds-only and sub-second ISO 8601 inputs (placeholder path), when called, then the strings are inserted as URL-encoded query parameters (``urlencode`` percent-encodes the ':' separators).""" from urllib.parse import quote_plus out = splunk.prepare_investigations_query(self.DEFAULT_QUERY, ts_min, ts_max, 10, 0) assert f"create_time_min={quote_plus(ts_min)}" in out assert f"create_time_max={quote_plus(ts_max)}" in out # --------------------------- to_mc_iso8601_utc --------------------------- class TestToMcIso8601Utc: """Given a timestamp string or datetime, when to_mc_iso8601_utc runs, then it returns the canonical Mission Control shape `YYYY-MM-DDTHH:MM:SS.ffffffZ` (always 6-digit microsecond precision).""" def test_given_get_fetch_time_window_format_when_normalized_then_z_suffix(self): """Given the format produced by get_fetch_time_window (e.g. '+00:00'), when normalized, then output ends with Z and preserves microseconds.""" out = splunk.to_mc_iso8601_utc("2025-12-03T11:53:45.138540+00:00") assert out == "2025-12-03T11:53:45.138540Z" def test_given_canonical_input_when_normalized_then_passes_through(self): """Given an already-canonical input (6-digit microseconds + Z), when normalized, then it passes through unchanged.""" ts = "2025-01-01T00:00:00.000000Z" assert splunk.to_mc_iso8601_utc(ts) == ts def test_given_negative_offset_when_normalized_then_converted_to_utc_z(self): """Given a `-05:00` offset, when normalized, then converted to UTC and emitted with Z.""" out = splunk.to_mc_iso8601_utc("2025-01-01T00:00:00-05:00") # 00:00 EST → 05:00 UTC assert out == "2025-01-01T05:00:00.000000Z" def test_given_subsecond_when_normalized_then_microseconds_preserved(self): """Given a sub-second offset input, when normalized, then microseconds are preserved.""" out = splunk.to_mc_iso8601_utc("2025-01-01T00:00:00.123456+00:00") assert out == "2025-01-01T00:00:00.123456Z" def test_given_no_subsecond_when_normalized_then_zero_microseconds_emitted(self): """Given an input without fractional seconds, when normalized, then zero-padded 6-digit microseconds are emitted (canonical shape is always preserved).""" out = splunk.to_mc_iso8601_utc("2025-01-01T00:00:00+00:00") assert out == "2025-01-01T00:00:00.000000Z" def test_given_naive_string_when_normalized_then_raises(self): """Given a naive ISO string (no tz), when normalized, then DemistoException is raised.""" with pytest.raises(splunk.DemistoException, match="naive"): splunk.to_mc_iso8601_utc("2025-01-01T00:00:00") def test_given_naive_datetime_when_normalized_then_raises(self): """Given a naive datetime, when normalized, then DemistoException is raised.""" with pytest.raises(splunk.DemistoException, match="naive"): splunk.to_mc_iso8601_utc(datetime(2025, 1, 1, 0, 0, 0)) def test_given_aware_datetime_when_normalized_then_canonical_z(self): """Given a tz-aware datetime, when normalized, then canonical Z output with zero-padded 6-digit microseconds.""" dt = datetime(2025, 1, 1, 12, 0, 0, tzinfo=UTC) assert splunk.to_mc_iso8601_utc(dt) == "2025-01-01T12:00:00.000000Z" # --------------------------- FetchHandlerFactory --------------------------- class TestFetchHandlerFactory: """Given a (possibly empty) selection of event types, when FetchHandlerFactory.build runs, then it returns the matching handlers in registration order, ignoring unknowns.""" def setup_method(self): # Snapshot the live registry so each test starts from a clean state and # can restore afterwards (factory is a class-level singleton). self._registry_snapshot = dict(splunk.FetchHandlerFactory._registry) def teardown_method(self): splunk.FetchHandlerFactory._registry = self._registry_snapshot def _stub(self, event_type: str): """Define a minimal concrete FetchHandler subclass usable in build().""" class _StubHandler(splunk.FetchHandler): def fetch(self, service, last_run, mapper, params): # noqa: D401 return splunk.FetchResult(incidents=[], last_run_delta={}) _StubHandler.event_type = event_type _StubHandler.__name__ = f"Stub{event_type}Handler" return _StubHandler def test_given_empty_registry_and_default_selection_then_returns_empty_list(self): """Given an empty registry, when build(None) runs (defaulting to ['Finding']), then no handlers are returned (Phase 3a default).""" splunk.FetchHandlerFactory._registry = {} assert splunk.FetchHandlerFactory.build(None) == [] def test_given_empty_registry_and_empty_selection_then_returns_empty_list(self): """Given an empty registry and an empty list, when build runs, then no handlers are returned.""" splunk.FetchHandlerFactory._registry = {} assert splunk.FetchHandlerFactory.build([]) == [] def test_given_investigation_only_when_built_then_single_handler(self): """Given a selection of ['Investigation'] and a registered Investigation stub, when build runs, then exactly one handler is returned.""" splunk.FetchHandlerFactory._registry = {} splunk.FetchHandlerFactory.register("Investigation", self._stub("Investigation")) out = splunk.FetchHandlerFactory.build(["Investigation"]) assert len(out) == 1 assert out[0].event_type == "Investigation" def test_given_both_selected_when_built_then_two_handlers_in_registration_order(self): """Given Finding registered first and Investigation second, when build(['Finding', 'Investigation']) runs, then handlers are returned in the requested selection order.""" splunk.FetchHandlerFactory._registry = {} splunk.FetchHandlerFactory.register("Finding", self._stub("Finding")) splunk.FetchHandlerFactory.register("Investigation", self._stub("Investigation")) out = splunk.FetchHandlerFactory.build(["Finding", "Investigation"]) assert [h.event_type for h in out] == ["Finding", "Investigation"] def test_given_unknown_type_when_built_then_silently_ignored(self): """Given an unknown event type alongside a known one, when build runs, then the unknown is silently ignored.""" splunk.FetchHandlerFactory._registry = {} splunk.FetchHandlerFactory.register("Finding", self._stub("Finding")) out = splunk.FetchHandlerFactory.build(["Finding", "Galaxy"]) assert [h.event_type for h in out] == ["Finding"] def test_given_default_with_finding_registered_then_returns_finding(self): """Given Finding is registered and selection is None, when build runs, then it falls back to ['Finding'] and returns one handler.""" splunk.FetchHandlerFactory._registry = {} splunk.FetchHandlerFactory.register("Finding", self._stub("Finding")) out = splunk.FetchHandlerFactory.build(None) assert len(out) == 1 assert out[0].event_type == "Finding" # --------------------------- parse_investigation / Investigation --------------------------- def _sample_investigation_row() -> dict: """Build a representative investigations row in-line (no fixture file dependency), reflecting the schema in plan §3.7.1.""" return { "investigation_guid": "5f4d3c2b-1a09-4b8c-9d7e-6f5a4b3c2d1e", "investigation_id": "ES-00015", "name": "Suspicious lateral movement", "description": "Multiple failed logons followed by privilege escalation.", "create_time": 1777446173.55, "update_time": 1777449999.12, "mc_create_time": 1777446173.55, "incident_origin": "MC Incident", "investigation_type": "default", "disposition": "disposition:6", "disposition_name": "Undetermined", "status": "1", "status_name": "New", "owner": "unassigned", "urgency": "high", "sensitivity": "Unassigned", "findings": {"incident_ids": ["F-1", "F-2"]}, "excluded_finding_ids": [], "implicit_finding_ids": ["IMP-1"], "intermediate_finding_ids": [], "consolidated_findings": {"summary": "n/a"}, "count_findings": 1, "risk_event_count": 0, "risk_score": 60.0, "risk_object": [], "risk_object_type": [], "src": ["192.168.0.2"], "dest": ["192.168.0.2"], "dvc": [], "orig_host": [], "src_user": ["unknown"], "user": ["tng\\crusher"], # backslash → must round-trip via json.dumps "is_investigation": True, "is_finding_group": False, "is_search_enriched": True, } class TestParseInvestigation: """Given an investigations row, when parse_investigation runs, then it returns a flattened dict tagged for the classifier.""" def test_given_row_when_parsed_then_event_type_tag_added(self): out = splunk.parse_investigation(_sample_investigation_row()) assert out[splunk.SPLUNK_ES_EVENT_TYPE_FIELD] == "Investigation" def test_given_row_when_parsed_then_findings_incident_ids_lifted(self): out = splunk.parse_investigation(_sample_investigation_row()) assert out["incident_ids"] == ["F-1", "F-2"] def test_given_row_when_parsed_then_risk_object_and_risk_object_type_distinct(self): row = _sample_investigation_row() row["risk_object"] = ["host-1"] row["risk_object_type"] = ["system"] out = splunk.parse_investigation(row) assert out["risk_object"] == ["host-1"] assert out["risk_object_type"] == ["system"] def test_given_row_when_parsed_then_user_and_src_user_remain_arrays(self): out = splunk.parse_investigation(_sample_investigation_row()) assert isinstance(out["user"], list) assert isinstance(out["src_user"], list) def test_given_row_when_parsed_then_input_not_mutated(self): row = _sample_investigation_row() original_keys = set(row.keys()) splunk.parse_investigation(row) assert set(row.keys()) == original_keys assert splunk.SPLUNK_ES_EVENT_TYPE_FIELD not in row def test_given_dotted_consolidated_findings_keys_when_parsed_then_collected_into_nested_object(self): """ Given: - A row whose `consolidated_findings` data is delivered as flat, dotted keys (e.g. `consolidated_findings.search_name`, `consolidated_findings.queue_id`, `consolidated_findings.dest`), which is how the investigations endpoint actually returns it. When: - parse_investigation is called. Then: - All dotted keys are removed from the top level. - Their values are nested under a single `consolidated_findings` key whose payload is a JSON string preserving the inner keys (without the prefix) and their original values. """ row = _sample_investigation_row() # Drop the legacy nested `consolidated_findings` so we exercise the # dotted-keys-only path explicitly. row.pop("consolidated_findings", None) row["consolidated_findings.search_name"] = ["Suspicious Login", "Brute Force"] row["consolidated_findings.queue_id"] = "None" row["consolidated_findings.dest"] = ["host-a", "host-b"] row["consolidated_findings.src_user"] = "unknown" out = splunk.parse_investigation(row) # No dotted keys remain at the top level. assert not any(k.startswith("consolidated_findings.") for k in out) # Single key holds the JSON-encoded nested payload. assert isinstance(out["consolidated_findings"], str) nested = json.loads(out["consolidated_findings"]) assert nested == { "search_name": ["Suspicious Login", "Brute Force"], "queue_id": "None", "dest": ["host-a", "host-b"], "src_user": "unknown", } def test_given_dotted_keys_and_nested_object_when_parsed_then_merged(self): """ Given: - A row that contains BOTH a pre-existing nested `consolidated_findings` object AND additional dotted-key entries (mixed schemas can occur during transition windows). When: - parse_investigation is called. Then: - The dotted keys are merged into the existing nested object (dotted keys win on conflict, since they reflect the latest top-level row state from Splunk), producing a single JSON string under `consolidated_findings`. """ row = _sample_investigation_row() row["consolidated_findings"] = {"queue_id": "old", "extra": "keep-me"} row["consolidated_findings.queue_id"] = "new" row["consolidated_findings.dest"] = ["host-a"] out = splunk.parse_investigation(row) nested = json.loads(out["consolidated_findings"]) # Dotted-key value overrides existing nested value on conflict. assert nested["queue_id"] == "new" # Pre-existing nested keys are retained. assert nested["extra"] == "keep-me" # Newly collected dotted key is present. assert nested["dest"] == ["host-a"] def test_given_consolidated_findings_object_when_parsed_then_serialized_to_json_string(self): """ Given: - A row whose `consolidated_findings` is a nested object containing both scalars (queue_id, src_user) and parallel array columns (search_name, _time, dest, risk_score), as returned by the investigations endpoint. When: - parse_investigation is called. Then: - The output's `consolidated_findings` is a JSON-encoded string (so the classifier can map it as-is into a longText incident field), and the original row remains untouched. """ row = _sample_investigation_row() payload = { "search_name": ["Suspicious Login", "Brute Force"], "_time": ["2025-01-01T10:00:00", "2025-01-01T10:05:00"], "dest": ["host-a", "host-b"], "risk_score": ["80", "90"], "queue_id": "None", "src_user": "unknown", } row["consolidated_findings"] = payload out = splunk.parse_investigation(row) assert isinstance(out["consolidated_findings"], str) assert json.loads(out["consolidated_findings"]) == payload # Source row must NOT be mutated. assert isinstance(row["consolidated_findings"], dict) def test_given_consolidated_findings_already_string_when_parsed_then_left_unchanged(self): """ Given: - A row whose `consolidated_findings` is already a JSON string (e.g. an upstream caller pre-serialized it). When: - parse_investigation is called. Then: - The string is preserved as-is (no double encoding). """ row = _sample_investigation_row() original_string = '{"queue_id":"None","src_user":"unknown"}' row["consolidated_findings"] = original_string out = splunk.parse_investigation(row) assert out["consolidated_findings"] == original_string def test_given_consolidated_findings_missing_when_parsed_then_no_key_added(self): """ Given: - A row where `consolidated_findings` is absent or None. When: - parse_investigation is called. Then: - No spurious key is materialized; absent stays absent. """ row = _sample_investigation_row() row.pop("consolidated_findings", None) out = splunk.parse_investigation(row) assert "consolidated_findings" not in out class TestInvestigationModel: """Given a parsed investigation row, when Investigation.to_incident runs, then it produces an XSOAR incident dict that satisfies the Phase 3a contract.""" def _mapper(self): m = MagicMock() m.should_map = False m.get_xsoar_user_by_splunk.side_effect = lambda u: u return m def test_given_happy_path_row_when_to_incident_then_no_type_set(self, mocker): mocker.patch.object(splunk.demisto, "params", return_value={"mirror_direction": "Incoming"}) mocker.patch.object(splunk.demisto, "integrationInstance", return_value="splunk_inst_1") inv = splunk.Investigation(splunk.parse_investigation(_sample_investigation_row())) incident = inv.to_incident(self._mapper()) assert "type" not in incident, "Phase 3a forbids the integration setting incident['type']" def test_given_happy_path_row_when_to_incident_then_event_type_in_rawjson(self, mocker): mocker.patch.object(splunk.demisto, "params", return_value={"mirror_direction": "None"}) mocker.patch.object(splunk.demisto, "integrationInstance", return_value="splunk_inst_1") inv = splunk.Investigation(splunk.parse_investigation(_sample_investigation_row())) incident = inv.to_incident(self._mapper()) raw = json.loads(incident["rawJSON"]) assert raw[splunk.SPLUNK_ES_EVENT_TYPE_FIELD] == "Investigation" def test_given_happy_path_row_when_to_incident_then_dbot_mirror_id_is_guid(self, mocker): mocker.patch.object(splunk.demisto, "params", return_value={"mirror_direction": "None"}) mocker.patch.object(splunk.demisto, "integrationInstance", return_value="splunk_inst_1") inv = splunk.Investigation(splunk.parse_investigation(_sample_investigation_row())) incident = inv.to_incident(self._mapper()) assert incident["dbotMirrorId"] == "5f4d3c2b-1a09-4b8c-9d7e-6f5a4b3c2d1e" assert incident["dbotMirrorInstance"] == "splunk_inst_1" assert incident["dbotMirrorDirection"] is None # MIRROR_DIRECTION["None"] is None def test_given_urgency_when_to_incident_then_severity_derived(self, mocker): mocker.patch.object(splunk.demisto, "params", return_value={}) mocker.patch.object(splunk.demisto, "integrationInstance", return_value="x") inv = splunk.Investigation(splunk.parse_investigation(_sample_investigation_row())) incident = inv.to_incident(self._mapper()) assert incident["severity"] == splunk.severity_to_level("high") def test_given_create_time_when_to_incident_then_occurred_is_rfc3339(self, mocker): mocker.patch.object(splunk.demisto, "params", return_value={}) mocker.patch.object(splunk.demisto, "integrationInstance", return_value="x") inv = splunk.Investigation(splunk.parse_investigation(_sample_investigation_row())) incident = inv.to_incident(self._mapper()) # RFC 3339 with timezone suffix. assert "T" in incident["occurred"] assert incident["occurred"].endswith("+00:00") or incident["occurred"].endswith("Z") def test_given_missing_optional_fields_when_to_incident_then_handled(self, mocker): mocker.patch.object(splunk.demisto, "params", return_value={}) mocker.patch.object(splunk.demisto, "integrationInstance", return_value="x") sparse_row = { "investigation_guid": "g-1", "investigation_id": "ES-1", "name": "n", "create_time": 1777446173.0, } inv = splunk.Investigation(splunk.parse_investigation(sparse_row)) incident = inv.to_incident(self._mapper()) assert incident["name"] == "n" assert "details" not in incident assert "severity" not in incident assert "owner" not in incident def test_given_backslash_in_user_when_to_incident_then_round_trips(self, mocker): mocker.patch.object(splunk.demisto, "params", return_value={}) mocker.patch.object(splunk.demisto, "integrationInstance", return_value="x") inv = splunk.Investigation(splunk.parse_investigation(_sample_investigation_row())) incident = inv.to_incident(self._mapper()) raw = json.loads(incident["rawJSON"]) assert raw["user"] == ["tng\\crusher"] def test_given_get_id_then_prefers_investigation_id(self): inv = splunk.Investigation(splunk.parse_investigation(_sample_investigation_row())) assert inv.get_id() == "ES-00015" def test_given_no_investigation_id_when_get_id_then_falls_back_to_guid(self): row = _sample_investigation_row() row.pop("investigation_id") inv = splunk.Investigation(splunk.parse_investigation(row)) assert inv.get_id() == "5f4d3c2b-1a09-4b8c-9d7e-6f5a4b3c2d1e" # ============================================================================ # Phase 3b — refactor + handlers + dispatcher # ============================================================================ class TestBuildFetchQueryGeneralized: """Given build_fetch_query is generalized, when called with the optional ``query_param_name``, then it reads the right param key with full BC.""" def test_given_no_override_then_reads_fetchQuery_BC(self): # GIVEN params = {"fetchQuery": "search index=notable"} # WHEN out = splunk.build_fetch_query(params) # THEN assert out == "search index=notable" def test_given_explicit_override_then_reads_that_key(self): # GIVEN params = { "fetchQuery": "search ignored", "investigations_fetch_query": '| rest "/foo"', } # WHEN out = splunk.build_fetch_query(params, query_param_name="investigations_fetch_query") # THEN assert out == '| rest "/foo"' def test_given_extract_fields_then_appended_to_chosen_query(self): # GIVEN params = { "investigations_fetch_query": '| rest "/foo"', "extractFields": "field_a,field_b", } # WHEN out = splunk.build_fetch_query(params, query_param_name="investigations_fetch_query") # THEN assert "| eval field_a=field_a" in out assert "| eval field_b=field_b" in out assert out.startswith('| rest "/foo"') class TestFetchFindingsRefactor: """Given fetch_findings has been refactored to return FetchResult, when called directly, then it does NOT call demisto.incidents/setLastRun and every produced incident is tagged splunk_es_event_type=Finding.""" def test_given_call_when_run_then_returns_FetchResult(self, mocker): # GIVEN mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mocker.patch("demistomock.getLastRun", return_value={"time": "2018-10-24T14:13:20"}) mocker.patch("demistomock.params", return_value={"fetchQuery": "something"}) incidents_mock = mocker.patch.object(demisto, "incidents") set_last_run_mock = mocker.patch.object(demisto, "setLastRun") mocker.patch("splunklib.results.JSONResultsReader", return_value=deepcopy(SAMPLE_RESPONSE)) service = Service("DONE") mapper = splunk.UserMappingObject(service, False) # WHEN result = splunk.fetch_findings(service=service, mapper=mapper) # THEN assert isinstance(result, splunk.FetchResult) assert isinstance(result.incidents, list) assert isinstance(result.last_run_delta, dict) # No side-effects — dispatcher owns those. incidents_mock.assert_not_called() set_last_run_mock.assert_not_called() def test_given_produced_incidents_when_inspected_then_tagged_with_Finding(self, mocker): # GIVEN mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mocker.patch("demistomock.getLastRun", return_value={"time": "2018-10-24T14:13:20"}) mocker.patch("demistomock.params", return_value={"fetchQuery": "something"}) mocker.patch.object(demisto, "incidents") mocker.patch.object(demisto, "setLastRun") mocker.patch("splunklib.results.JSONResultsReader", return_value=deepcopy(SAMPLE_RESPONSE)) service = Service("DONE") mapper = splunk.UserMappingObject(service, False) # WHEN result = splunk.fetch_findings(service=service, mapper=mapper) # THEN assert result.incidents, "expected at least one produced incident" for inc in result.incidents: raw = json.loads(inc["rawJSON"]) assert raw[splunk.SPLUNK_ES_EVENT_TYPE_FIELD] == "Finding" # Integration must NOT set incident["type"]; the Classifier owns it. assert "type" not in inc def test_given_FetchResult_when_inspected_then_last_run_delta_has_BC_keys(self, mocker): # GIVEN mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mocker.patch("demistomock.getLastRun", return_value={"time": "2018-10-24T14:13:20"}) mocker.patch("demistomock.params", return_value={"fetchQuery": "something"}) mocker.patch.object(demisto, "incidents") mocker.patch.object(demisto, "setLastRun") mocker.patch("splunklib.results.JSONResultsReader", return_value=deepcopy(SAMPLE_RESPONSE)) service = Service("DONE") mapper = splunk.UserMappingObject(service, False) # WHEN result = splunk.fetch_findings(service=service, mapper=mapper) # THEN — same top-level keys as the legacy setLastRun payload. for key in ("next_run_earliest_time", "offset", "next_run_found_incidents_ids"): assert key in result.last_run_delta class TestFindingsFetchHandler: """Given FindingsFetchHandler wraps fetch_findings, when invoked, then output is identical to calling fetch_findings directly (in the non-enrichment branch).""" def setup_method(self): # Snapshot/restore the registry so factory mutations don't leak. self._registry_snapshot = dict(splunk.FetchHandlerFactory._registry) def teardown_method(self): splunk.FetchHandlerFactory._registry = self._registry_snapshot def test_given_default_construction_then_event_type_and_last_run_key_set(self): h = splunk.FindingsFetchHandler() assert h.event_type == "Finding" assert h.last_run_key is None def test_given_no_enrichments_when_fetch_then_passes_through_to_fetch_findings(self, mocker): # GIVEN mocker.patch.object(splunk, "ENABLED_ENRICHMENTS", []) mocker.patch.object(splunk, "get_current_splunk_time", return_value="2018-10-24T14:13:20.000+00:00") mocker.patch("demistomock.getLastRun", return_value={"time": "2018-10-24T14:13:20"}) mocker.patch("demistomock.params", return_value={"fetchQuery": "something"}) mocker.patch("splunklib.results.JSONResultsReader", return_value=deepcopy(SAMPLE_RESPONSE)) service = Service("DONE") mapper = splunk.UserMappingObject(service, False) handler = splunk.FindingsFetchHandler() # WHEN result = handler.fetch(service, last_run={}, mapper=mapper, params=demisto.params()) # THEN assert isinstance(result, splunk.FetchResult) for inc in result.incidents: raw = json.loads(inc["rawJSON"]) assert raw[splunk.SPLUNK_ES_EVENT_TYPE_FIELD] == "Finding" class _FakeReader: """Minimal stand-in for splunklib.results.JSONResultsReader — yields rows.""" def __init__(self, rows): self._rows = rows def __iter__(self): return iter(self._rows) class TestInvestigationsFetchHandler: """Given InvestigationsFetchHandler runs against mocked oneshot results, when fetched, then incidents are tagged Investigation, dedup is honored, pagination cursor advances/resets correctly, and enrichment is NOT called.""" def setup_method(self): self._registry_snapshot = dict(splunk.FetchHandlerFactory._registry) def teardown_method(self): splunk.FetchHandlerFactory._registry = self._registry_snapshot def _service_returning(self, rows, mocker): """Build a mock service whose service.jobs.oneshot returns ``rows``.""" service = mocker.MagicMock() service.jobs.oneshot.return_value = MagicMock() mocker.patch("splunklib.results.JSONResultsReader", return_value=_FakeReader(rows)) return service def _mapper(self): m = MagicMock() m.should_map = False m.get_xsoar_user_by_splunk.side_effect = lambda u: u return m def _common_setup(self, mocker, params_extra=None): params = { "investigations_fetch_query": ( '| rest "/servicesNS/nobody/missioncontrol/public/v2/investigations' '?search_format=true&FETCH_FILTER_PLACEHOLDER"' ), "investigations_max_fetch": "2", "first_fetch": "7 days", "investigations_first_fetch": "7 days", "mirror_direction": "None", } if params_extra: params.update(params_extra) mocker.patch.object(demisto, "params", return_value=params) mocker.patch.object(demisto, "integrationInstance", return_value="splunk_inst_1") # Defensive: silence demisto.error so the conftest no-stdout fixture is # satisfied if a handler-level exception is logged. mocker.patch.object(demisto, "error") mocker.patch.object( splunk, "get_fetch_time_window", return_value=( "2025-01-01T00:00:00.000000+00:00", "2025-01-08T00:00:00.000000+00:00", ), ) return params def test_given_rows_when_fetch_then_all_incidents_tagged_Investigation(self, mocker): # GIVEN params = self._common_setup(mocker) rows = [_sample_investigation_row()] rows[0]["investigation_id"] = "ES-001" service = self._service_returning(rows, mocker) handler = splunk.InvestigationsFetchHandler() # WHEN result = handler.fetch(service, last_run={}, mapper=self._mapper(), params=params) # THEN assert len(result.incidents) == 1 raw = json.loads(result.incidents[0]["rawJSON"]) assert raw[splunk.SPLUNK_ES_EVENT_TYPE_FIELD] == "Investigation" assert result.incidents[0]["dbotMirrorId"] == raw["investigation_guid"] # Integration MUST NOT set incident["type"]. assert "type" not in result.incidents[0] def test_given_dup_id_in_last_run_when_fetch_then_dropped(self, mocker): # GIVEN params = self._common_setup(mocker) row_a = _sample_investigation_row() row_a["investigation_id"] = "ES-001" row_b = deepcopy(_sample_investigation_row()) row_b["investigation_id"] = "ES-002" service = self._service_returning([row_a, row_b], mocker) handler = splunk.InvestigationsFetchHandler() # `remove_irrelevant_incident_ids` parses occurred_time using # ISO_FORMAT_TZ_AWARE (`%Y-%m-%dT%H:%M:%S.%f%z`) — must use the +00:00 # offset shape, NOT the Mission Control Z-suffix shape. last_run = {"found_incidents_ids": {"ES-001": {"occurred_time": "2025-01-08T00:00:00.000000+00:00"}}} # WHEN result = handler.fetch(service, last_run=last_run, mapper=self._mapper(), params=params) # THEN assert len(result.incidents) == 1 raw = json.loads(result.incidents[0]["rawJSON"]) assert raw["investigation_id"] == "ES-002" def test_given_full_page_when_fetch_then_offset_advances(self, mocker): # GIVEN limit=2 and 2 rows returned params = self._common_setup(mocker, {"investigations_max_fetch": "2"}) rows = [ {**_sample_investigation_row(), "investigation_id": "ES-100"}, {**_sample_investigation_row(), "investigation_id": "ES-101"}, ] service = self._service_returning(rows, mocker) handler = splunk.InvestigationsFetchHandler() # WHEN result = handler.fetch(service, last_run={"offset": 0}, mapper=self._mapper(), params=params) # THEN assert result.last_run_delta["offset"] == 2 def test_given_partial_page_when_fetch_then_offset_resets(self, mocker): # GIVEN limit=5 and only 1 row returned params = self._common_setup(mocker, {"investigations_max_fetch": "5"}) rows = [{**_sample_investigation_row(), "investigation_id": "ES-200"}] service = self._service_returning(rows, mocker) handler = splunk.InvestigationsFetchHandler() # WHEN result = handler.fetch(service, last_run={"offset": 5}, mapper=self._mapper(), params=params) # THEN assert result.last_run_delta["offset"] == 0 def test_given_fetch_when_run_then_run_enrichment_mechanism_NOT_called(self, mocker): # GIVEN params = self._common_setup(mocker) rows = [{**_sample_investigation_row(), "investigation_id": "ES-300"}] service = self._service_returning(rows, mocker) enrich_mock = mocker.patch.object(splunk, "run_enrichment_mechanism") handler = splunk.InvestigationsFetchHandler() # WHEN handler.fetch(service, last_run={}, mapper=self._mapper(), params=params) # THEN enrich_mock.assert_not_called() def test_given_fetch_when_run_then_event_type_injected_once(self, mocker): # GIVEN params = self._common_setup(mocker) row = _sample_investigation_row() row["investigation_id"] = "ES-400" service = self._service_returning([row], mocker) handler = splunk.InvestigationsFetchHandler() # WHEN result = handler.fetch(service, last_run={}, mapper=self._mapper(), params=params) # THEN — exactly one occurrence of the tag in rawJSON. raw_str = result.incidents[0]["rawJSON"] assert raw_str.count(f'"{splunk.SPLUNK_ES_EVENT_TYPE_FIELD}"') == 1 def test_given_rows_with_guid_when_fetch_then_enrich_with_splunk_notes_called(self, mocker): """Given fetched rows with `investigation_guid`, when the handler runs, then `enrich_with_splunk_notes_v2` is called with a guid-keyed map and `is_fetch=True` (mirrors the Findings enrichment pattern; status doc Phase 3b, lines 1562–1567). The handler was switched from the legacy ``enrich_with_splunk_notes`` helper to the v2 KV-store variant; this test asserts the new contract. """ # GIVEN params = self._common_setup(mocker) row = _sample_investigation_row() row["investigation_id"] = "ES-NOTES-1" service = self._service_returning([row], mocker) enrich_spy = mocker.patch.object(splunk, "enrich_with_splunk_notes_v2") handler = splunk.InvestigationsFetchHandler() # WHEN handler.fetch(service, last_run={}, mapper=self._mapper(), params=params) # THEN — called exactly once, with the guid as the dict key and is_fetch=True. enrich_spy.assert_called_once() call_args = enrich_spy.call_args # Positional args: (service, guid_to_row_map) passed_map = call_args.args[1] assert row["investigation_guid"] in passed_map assert passed_map[row["investigation_guid"]] is row # is_fetch must be True for the fetch path (vs. the modified-remote-data path). assert call_args.kwargs.get("is_fetch") is True # v2 helper additionally takes a `last_update_splunk_timestamp` epoch float # so it can scope the KV-store query by `update_time`. assert "last_update_splunk_timestamp" in call_args.kwargs assert isinstance(call_args.kwargs["last_update_splunk_timestamp"], float) def test_given_handler_failure_when_fetch_then_logs_error_and_reraises(self, mocker): """Given the inner `_do_fetch` raises, when `fetch` runs, then `demisto.error` is called with the handler+event-type prefix and the exception is re-raised (plan §3.10 error-path contract).""" # GIVEN self._common_setup(mocker) boom = RuntimeError("simulated downstream failure") mocker.patch.object(splunk.InvestigationsFetchHandler, "_do_fetch", side_effect=boom) # The common setup already patches demisto.error; recapture it as a spy. error_spy = mocker.patch.object(demisto, "error") handler = splunk.InvestigationsFetchHandler() # WHEN / THEN — exception re-raised. with pytest.raises(RuntimeError, match="simulated downstream failure"): handler.fetch(service=MagicMock(), last_run={}, mapper=self._mapper(), params={}) # AND — error was logged with the handler/event-type prefix. error_spy.assert_called_once() logged = error_spy.call_args.args[0] assert "InvestigationsFetchHandler" in logged assert "Investigation" in logged assert "simulated downstream failure" in logged def test_given_first_fetch_when_fetch_then_investigations_first_fetch_used_without_mutating_params(self, mocker): """Given an empty last_run (first-fetch path), when the handler runs, then `get_fetch_time_window` receives a params copy whose `first_fetch` is the value of `investigations_first_fetch`, and the caller's `params` dict is NOT mutated (status doc Phase 3b lines 1520–1523: `params_for_window = dict(params, first_fetch=...)`).""" # GIVEN params = { "investigations_fetch_query": ( '| rest "/servicesNS/nobody/missioncontrol/public/v2/investigations' '?search_format=true&FETCH_FILTER_PLACEHOLDER"' ), "investigations_max_fetch": "2", "investigations_first_fetch": "30 days", "first_fetch": "10 minutes", # Findings-side default — must NOT leak into investigations window "mirror_direction": "None", } original_params_snapshot = deepcopy(params) mocker.patch.object(demisto, "params", return_value=params) mocker.patch.object(demisto, "integrationInstance", return_value="splunk_inst_1") mocker.patch.object(demisto, "error") time_window_spy = mocker.patch.object( splunk, "get_fetch_time_window", return_value=( "2025-01-01T00:00:00.000000+00:00", "2025-01-08T00:00:00.000000+00:00", ), ) service = self._service_returning([], mocker) handler = splunk.InvestigationsFetchHandler() # WHEN — empty last_run triggers the first-fetch path. handler.fetch(service, last_run={}, mapper=self._mapper(), params=params) # THEN — get_fetch_time_window saw `first_fetch == investigations_first_fetch`. time_window_spy.assert_called_once() passed_params = time_window_spy.call_args.args[0] assert passed_params["first_fetch"] == "30 days" # AND — caller's params dict was not mutated; the Findings-side `first_fetch` survives. assert params == original_params_snapshot assert params["first_fetch"] == "10 minutes" def test_given_full_page_when_fetch_then_delta_keeps_window_with_all_5_keys(self, mocker): """Given a full page (advance branch), when the handler runs, then the 5-key last-run delta is returned and the window is kept (`time == ts_min`, `next_run_latest_time == ts_max`) — see status doc Phase 3b lines 1618–1631 for the cursor-advance contract.""" # GIVEN limit=2, exactly 2 rows -> next_offset advances, window held. params = self._common_setup(mocker, {"investigations_max_fetch": "2"}) rows = [ {**_sample_investigation_row(), "investigation_id": "ES-501"}, {**_sample_investigation_row(), "investigation_id": "ES-502"}, ] service = self._service_returning(rows, mocker) handler = splunk.InvestigationsFetchHandler() # WHEN result = handler.fetch(service, last_run={"offset": 0}, mapper=self._mapper(), params=params) # THEN — all 5 keys present. delta = result.last_run_delta for key in ("time", "next_run_earliest_time", "next_run_latest_time", "offset", "found_incidents_ids"): assert key in delta, f"missing key={key} in last_run_delta" # AND — `time` is an alias of `next_run_earliest_time` for cursor-reader BC. assert delta["time"] == delta["next_run_earliest_time"] # AND — window held: `next_run_earliest_time` == ts_min (canonical Z), latest == ts_max. assert delta["next_run_earliest_time"] == "2025-01-01T00:00:00.000000Z" assert delta["next_run_latest_time"] == "2025-01-08T00:00:00.000000Z" assert delta["offset"] == 2 def test_given_partial_page_when_fetch_then_delta_advances_window_and_clears_latest(self, mocker): """Given a partial page (reset branch), when the handler runs, then the window advances (`time == ts_max`, `next_run_latest_time is None`) and offset resets — status doc Phase 3b lines 1618–1623.""" # GIVEN limit=5, only 1 row -> next_offset resets, window advances. params = self._common_setup(mocker, {"investigations_max_fetch": "5"}) rows = [{**_sample_investigation_row(), "investigation_id": "ES-RESET-1"}] service = self._service_returning(rows, mocker) handler = splunk.InvestigationsFetchHandler() # WHEN result = handler.fetch(service, last_run={"offset": 0}, mapper=self._mapper(), params=params) # THEN delta = result.last_run_delta assert delta["offset"] == 0 assert delta["next_run_latest_time"] is None # Window advanced to ts_max. assert delta["next_run_earliest_time"] == "2025-01-08T00:00:00.000000Z" assert delta["time"] == delta["next_run_earliest_time"] class TestFetchIncidentsDispatcher: """Given the refactored fetch_incidents dispatcher, when called with different ``fetch_event_types`` configurations, then the right handlers run, last-run namespacing is correct, and BC defaults preserve old behavior.""" def setup_method(self): self._registry_snapshot = dict(splunk.FetchHandlerFactory._registry) def teardown_method(self): splunk.FetchHandlerFactory._registry = self._registry_snapshot def _install_stub(self, event_type: str, last_run_key, incidents=None, delta=None): """Install a stub handler that returns a deterministic FetchResult.""" captured: dict = {} class _StubHandler(splunk.FetchHandler): def fetch(self, service, last_run, mapper, params): # noqa: D401 captured["service"] = service captured["last_run"] = last_run captured["mapper"] = mapper captured["params"] = params return splunk.FetchResult( incidents=incidents or [], last_run_delta=delta or {}, ) _StubHandler.event_type = event_type _StubHandler.last_run_key = last_run_key _StubHandler.__name__ = f"Stub{event_type}Handler" splunk.FetchHandlerFactory.register(event_type, _StubHandler) return captured def test_given_unset_fetch_event_types_then_only_findings_runs(self, mocker): # GIVEN — only Findings stub registered, default fetch_event_types absent splunk.FetchHandlerFactory._registry = {} finding_inc = {"name": "f", "rawJSON": json.dumps({"splunk_es_event_type": "Finding"})} finding_capture = self._install_stub("Finding", None, [finding_inc], {"next_run_earliest_time": "2025-01-08"}) mocker.patch.object(demisto, "params", return_value={"fetchQuery": "x"}) mocker.patch.object(demisto, "getLastRun", return_value={"time": "2025-01-01"}) incidents_mock = mocker.patch.object(demisto, "incidents") set_last_run_mock = mocker.patch.object(demisto, "setLastRun") # WHEN splunk.fetch_incidents(service=MagicMock(), mapper=MagicMock()) # THEN incidents_mock.assert_called_once_with([finding_inc]) last_run = set_last_run_mock.call_args[0][0] assert last_run["time"] == "2025-01-01" # carried over assert last_run["next_run_earliest_time"] == "2025-01-08" # top-level merge (BC) assert "investigations" not in last_run assert finding_capture["last_run"] == {"time": "2025-01-01"} def test_given_investigation_only_then_only_investigations_runs_and_namespaced(self, mocker): # GIVEN splunk.FetchHandlerFactory._registry = {} inv_inc = {"name": "i", "rawJSON": json.dumps({"splunk_es_event_type": "Investigation"})} inv_capture = self._install_stub("Investigation", "investigations", [inv_inc], {"offset": 50, "time": "2025-01-08"}) # Also register Finding so unknown filtering doesn't leak. self._install_stub("Finding", None, [], {}) mocker.patch.object(demisto, "params", return_value={"fetch_event_types": "Investigation"}) mocker.patch.object( demisto, "getLastRun", return_value={"time": "2024-12-25", "investigations": {"offset": 0}}, ) incidents_mock = mocker.patch.object(demisto, "incidents") set_last_run_mock = mocker.patch.object(demisto, "setLastRun") # WHEN splunk.fetch_incidents(service=MagicMock(), mapper=MagicMock()) # THEN incidents_mock.assert_called_once_with([inv_inc]) last_run = set_last_run_mock.call_args[0][0] # Top-level keys untouched. assert last_run["time"] == "2024-12-25" # Namespaced keys merged. assert last_run["investigations"]["offset"] == 50 assert last_run["investigations"]["time"] == "2025-01-08" # Handler received only the namespaced scope. assert inv_capture["last_run"] == {"offset": 0} def test_given_both_selected_then_both_handlers_run_and_merged(self, mocker): # GIVEN splunk.FetchHandlerFactory._registry = {} finding_inc = {"name": "f", "rawJSON": json.dumps({"splunk_es_event_type": "Finding"})} inv_inc = {"name": "i", "rawJSON": json.dumps({"splunk_es_event_type": "Investigation"})} self._install_stub("Finding", None, [finding_inc], {"next_run_earliest_time": "F"}) self._install_stub("Investigation", "investigations", [inv_inc], {"offset": 100}) mocker.patch.object( demisto, "params", return_value={"fetch_event_types": "Finding,Investigation"}, ) mocker.patch.object(demisto, "getLastRun", return_value={}) incidents_mock = mocker.patch.object(demisto, "incidents") set_last_run_mock = mocker.patch.object(demisto, "setLastRun") # WHEN splunk.fetch_incidents(service=MagicMock(), mapper=MagicMock()) # THEN — Findings before Investigation per registration order. emitted = incidents_mock.call_args[0][0] assert emitted == [finding_inc, inv_inc] last_run = set_last_run_mock.call_args[0][0] assert last_run["next_run_earliest_time"] == "F" assert last_run["investigations"]["offset"] == 100 def test_given_argToList_handles_csv_string(self, mocker): # GIVEN — argToList accepts a comma-separated string. splunk.FetchHandlerFactory._registry = {} finding_inc = {"name": "f", "rawJSON": json.dumps({"splunk_es_event_type": "Finding"})} inv_inc = {"name": "i", "rawJSON": json.dumps({"splunk_es_event_type": "Investigation"})} self._install_stub("Finding", None, [finding_inc], {}) self._install_stub("Investigation", "investigations", [inv_inc], {}) mocker.patch.object( demisto, "params", return_value={"fetch_event_types": "Finding, Investigation"}, ) mocker.patch.object(demisto, "getLastRun", return_value={}) incidents_mock = mocker.patch.object(demisto, "incidents") mocker.patch.object(demisto, "setLastRun") # WHEN splunk.fetch_incidents(service=MagicMock(), mapper=MagicMock()) # THEN emitted = incidents_mock.call_args[0][0] assert len(emitted) == 2 def test_given_handler_raises_when_dispatcher_runs_then_logs_error_and_reraises(self, mocker): """Given a handler raises mid-fetch, when the dispatcher runs, then `demisto.error` is called with the handler/event-type prefix and the exception is re-raised; no `demisto.incidents`/`setLastRun` emit happens (plan §3.10 + status doc Phase 3b deviation #4).""" # GIVEN splunk.FetchHandlerFactory._registry = {} boom = RuntimeError("dispatcher-level boom") class _RaisingHandler(splunk.FetchHandler): event_type = "Investigation" last_run_key = "investigations" def fetch(self, service, last_run, mapper, params): raise boom splunk.FetchHandlerFactory.register("Investigation", _RaisingHandler) mocker.patch.object(demisto, "params", return_value={"fetch_event_types": "Investigation"}) mocker.patch.object(demisto, "getLastRun", return_value={}) incidents_mock = mocker.patch.object(demisto, "incidents") set_last_run_mock = mocker.patch.object(demisto, "setLastRun") error_spy = mocker.patch.object(demisto, "error") # WHEN / THEN with pytest.raises(RuntimeError, match="dispatcher-level boom"): splunk.fetch_incidents(service=MagicMock(), mapper=MagicMock()) # AND — error logged with handler name + event_type, no emit happened. error_spy.assert_called_once() logged = error_spy.call_args.args[0] assert "_RaisingHandler" in logged assert "Investigation" in logged assert "dispatcher-level boom" in logged incidents_mock.assert_not_called() set_last_run_mock.assert_not_called() class TestListModifiedInvestigations: """Given an investigations endpoint response, when list_modified_investigations runs, then it returns the row dicts (or an empty list on failure).""" @staticmethod def _service_returning(payload) -> MagicMock: """Build a fake `client.Service` whose `.get(endpoint)` returns ``payload`` wrapped in the splunklib body/read shape. ``payload`` may be either the raw list of rows (matching the v2 contract that `_fetch_modified_investigations_page` consumes) or a ``{"entry": [...]}`` wrapper kept for back-compat with older fixtures — the wrapper is unwrapped automatically. """ if isinstance(payload, dict) and "entry" in payload: payload = payload["entry"] service = MagicMock() body = MagicMock() body.read.return_value = json.dumps(payload).encode("utf-8") response = MagicMock() response.body = body service.get.return_value = response return service def test_given_two_rows_when_called_then_both_returned_and_url_carries_update_time_min(self): """ Given: - The v2 endpoint returns two rows with `investigation_guid` values g-1 and g-2. When: - list_modified_investigations is called with update_time_min="2026-04-29T10:00:00Z". Then: - Both rows are returned, and the URL hit on the service contains `update_time_min=2026-04-29T10:00:00Z`. """ service = self._service_returning({"entry": [{"investigation_guid": "g-1"}, {"investigation_guid": "g-2"}]}) result = splunk.list_modified_investigations(service, update_time_min="2026-04-29T10:00:00Z") assert [r["investigation_guid"] for r in result] == ["g-1", "g-2"] # `update_time_min` is forwarded as a kwarg on `service.get(...)`, # not baked into the URL — splunklib serialises kwargs into the query # string at HTTP-send time. assert service.get.call_args.kwargs["update_time_min"] == "2026-04-29T10:00:00Z" def test_given_row_missing_guid_when_called_then_id_row_is_kept(self): """ Given: - The endpoint returns a row that has only `investigation_id`. When: - list_modified_investigations runs. Then: - The row is kept in the returned list (the helper does not drop it). """ service = self._service_returning({"entry": [{"investigation_id": "id-only"}]}) result = splunk.list_modified_investigations(service, update_time_min="2026-04-29T10:00:00Z") assert len(result) == 1 assert result[0]["investigation_id"] == "id-only" def test_given_row_missing_both_when_called_then_silently_skipped(self): """ Given: - The endpoint returns a row that has neither `investigation_guid` nor `investigation_id`. When: - list_modified_investigations runs. Then: - That row is silently skipped (not present in the result). """ service = self._service_returning({"entry": [{"unrelated_field": "value"}, {"investigation_guid": "g-keep"}]}) result = splunk.list_modified_investigations(service, update_time_min="2026-04-29T10:00:00Z") assert len(result) == 1 assert result[0].get("investigation_guid") == "g-keep" def test_given_caller_page_size_999_when_called_then_url_carries_limit_100(self): """ Given: - A caller passes page_size=999. When: - list_modified_investigations runs. Then: - The URL hit on the service contains `limit=100` (clamped to INVESTIGATIONS_MAX_LIMIT). """ service = self._service_returning({"entry": []}) splunk.list_modified_investigations(service, update_time_min="2026-04-29T10:00:00Z", page_size=999) # `limit` is forwarded as a kwarg on `service.get(...)`; assert it was # clamped to INVESTIGATIONS_MAX_LIMIT (100) before being passed on. assert service.get.call_args.kwargs["limit"] == 100 def test_given_full_page_when_called_then_offset_advances_and_short_page_stops_loop(self, mocker): """ Given: - The first page returns a full page of rows (page_size rows) so the loop must request the next page; the second page is short, ending pagination. When: - list_modified_investigations runs with a small page_size for clarity. Then: - Two service.get calls are made; the first carries `offset=0`, the second `offset=`. All rows from both pages are returned. """ page_size = 2 first_page = [{"investigation_guid": "g-1"}, {"investigation_guid": "g-2"}] second_page = [{"investigation_guid": "g-3"}] # short page → loop stops # Patch the per-page fetch helper directly so the test does not need to # juggle multi-call MagicMock side_effects on `service.get`. page_spy = mocker.patch.object(splunk, "_fetch_modified_investigations_page", side_effect=[first_page, second_page]) result = splunk.list_modified_investigations( MagicMock(), update_time_min="2026-04-29T10:00:00Z", page_size=page_size, max_total=10 ) assert page_spy.call_count == 2 assert page_spy.call_args_list[0].args[3] == 0 # first offset assert page_spy.call_args_list[1].args[3] == page_size # advanced offset assert [r["investigation_guid"] for r in result] == ["g-1", "g-2", "g-3"] def test_given_max_total_reached_when_paginating_then_loop_stops_at_cap(self, mocker): """ Given: - Every page is full (so the loop would otherwise keep going). When: - list_modified_investigations runs with max_total=3 and page_size=2. Then: - Exactly 3 rows are returned and no further pages are requested once the cap is reached. """ full_page = [{"investigation_guid": "x"}, {"investigation_guid": "y"}] # Each call returns a fresh page of two distinct ids (so dedup does not interfere). def _page(_svc, _t, _ps, offset): return [ {"investigation_guid": f"g-{offset}-a"}, {"investigation_guid": f"g-{offset}-b"}, ] page_spy = mocker.patch.object(splunk, "_fetch_modified_investigations_page", side_effect=_page) result = splunk.list_modified_investigations( MagicMock(), update_time_min="2026-04-29T10:00:00Z", page_size=2, max_total=3 ) assert len(result) == 3 # 2 calls suffice: page 1 yields 2 rows, page 2 yields 1 more before max_total hits. assert page_spy.call_count == 2 # `full_page` reference is intentionally unused; kept for readability only. _ = full_page def test_given_endpoint_raises_when_called_then_returns_empty_and_logs_error(self, mocker): """ Given: - The service.get call raises an exception. When: - list_modified_investigations runs. Then: - The function returns an empty list and demisto.error is called once, so a single endpoint hiccup does not break Findings mirror-in. """ service = MagicMock() service.get.side_effect = RuntimeError("boom") error_spy = mocker.patch.object(demisto, "error") result = splunk.list_modified_investigations(service, update_time_min="2026-04-29T10:00:00Z") assert result == [] assert error_spy.call_count == 1 def test_given_mapper_with_mapping_enabled_when_called_then_owner_is_translated_to_xsoar_user(self): """ Given: - The v2 endpoint returns a row with `owner` = "splunk_user". - A `UserMappingObject` whose `should_map` is True and that resolves "splunk_user" → "xsoar_user". When: - list_modified_investigations is called with `mapper=`. Then: - The returned row's `owner` field is rewritten to "xsoar_user", matching the Findings owner-mapping behaviour in get_modified_remote_data_command. """ service = self._service_returning({"entry": [{"investigation_guid": "g-1", "owner": "splunk_user"}]}) mapper = MagicMock() mapper.should_map = True mapper.get_xsoar_user_by_splunk.return_value = "xsoar_user" # Delegate to the real helper so we exercise the same code path Findings use. mapper.map_owner_to_xsoar_user.side_effect = lambda rows: splunk.UserMappingObject.map_owner_to_xsoar_user(mapper, rows) result = splunk.list_modified_investigations(service, update_time_min="2026-04-29T10:00:00Z", mapper=mapper) assert result[0]["owner"] == "xsoar_user" mapper.get_xsoar_user_by_splunk.assert_called_once_with("splunk_user") def test_given_mapper_with_mapping_disabled_when_called_then_owner_is_not_modified(self): """ Given: - The v2 endpoint returns a row with `owner` = "splunk_user". - A `UserMappingObject` whose `should_map` is False (mapping disabled). When: - list_modified_investigations is called with `mapper=`. Then: - The row's `owner` is left untouched (no mapping lookup performed). """ service = self._service_returning({"entry": [{"investigation_guid": "g-1", "owner": "splunk_user"}]}) mapper = MagicMock() mapper.should_map = False mapper.map_owner_to_xsoar_user.side_effect = lambda rows: splunk.UserMappingObject.map_owner_to_xsoar_user(mapper, rows) result = splunk.list_modified_investigations(service, update_time_min="2026-04-29T10:00:00Z", mapper=mapper) assert result[0]["owner"] == "splunk_user" mapper.get_xsoar_user_by_splunk.assert_not_called() def test_given_no_mapper_when_called_then_owner_is_returned_as_is(self): """ Given: - The v2 endpoint returns a row with `owner` = "splunk_user". - No mapper is supplied (caller did not pass one). When: - list_modified_investigations is called without `mapper`. Then: - The owner value is returned untouched, preserving the previous behaviour for callers that have not opted into user mapping. """ service = self._service_returning({"entry": [{"investigation_guid": "g-1", "owner": "splunk_user"}]}) result = splunk.list_modified_investigations(service, update_time_min="2026-04-29T10:00:00Z") assert result[0]["owner"] == "splunk_user" def test_given_full_v2_row_when_called_then_rows_returned_in_canonical_parsed_shape(self): """ Given: - The v2 endpoint returns a representative investigations row (`_sample_investigation_row`) carrying: * a nested `findings.incident_ids` array, * a nested `consolidated_findings` dict, * none of the fetch-time normalisations applied. When: - list_modified_investigations is called. Then: - Every returned row carries the canonical fetch-time shape produced by `parse_investigation`: * `splunk_es_event_type == "Investigation"` (classifier tag), * `incident_ids` lifted to top level, * `consolidated_findings` JSON-serialised to a string. - This guarantees mirror-in payloads match what the original fetch flow emits via the Investigation class, so the classifier/mapper receive consistently-shaped data on both code paths. """ service = self._service_returning([_sample_investigation_row()]) result = splunk.list_modified_investigations(service, update_time_min="2026-04-29T10:00:00Z") assert len(result) == 1 parsed_row = result[0] # Classifier routing tag stamped by parse_investigation. assert parsed_row[splunk.SPLUNK_ES_EVENT_TYPE_FIELD] == "Investigation" # `findings.incident_ids` lifted to top level. assert parsed_row["incident_ids"] == ["F-1", "F-2"] # `consolidated_findings` serialised to JSON string (vs the dict in the raw row). assert isinstance(parsed_row["consolidated_findings"], str) assert json.loads(parsed_row["consolidated_findings"]) == {"summary": "n/a"} class TestGetModifiedRemoteDataInvestigations: """Given fetch_event_types containing (or not containing) Investigation, when get_modified_remote_data_command runs, then list_modified_investigations is invoked (or skipped) and its rows are appended to the response.""" def _build_kwargs(self, mocker) -> dict: service = mocker.patch.object(client, "Service") return { "service": service, "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": True, "close_end_statuses": True, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(service, False), } def test_given_investigations_selected_when_command_runs_then_guid_rows_appended(self, mocker): """ Given: - `fetch_event_types` is "Finding,Investigation". - The Findings audit-log SPL search yields ids F-1 and F-2. - list_modified_investigations is mocked to return [{"investigation_guid": "g-1"}]. When: - get_modified_remote_data_command runs. Then: - The SplunkGetModifiedRemoteDataResponse carries all three rows (F-1, F-2 from Findings + g-1 from Investigations) so the platform can route each to its own get-remote-data handler. """ kwargs = self._build_kwargs(mocker) mocker.patch.object(demisto, "params", return_value={"timezone": "0", "fetch_event_types": "Finding,Investigation"}) finding_rows = [ {"rule_id": "F-1", "event_id": "F-1", "review_time": "1737547610.56"}, {"rule_id": "F-2", "event_id": "F-2", "review_time": "1737547611.56"}, ] mocker.patch("SplunkPyV2.results.JSONResultsReader", return_value=finding_rows) mocker.patch("SplunkPyV2.get_current_splunk_time", return_value="2021-02-09T17:41:30.589575+02:00") # The mirror-in dedup loop in `get_modified_remote_data_command` filters # out any investigation row missing `investigation_guid` OR `update_time` # (these together form the dedup cache key). Provide both so the row # survives the dedup filter and is appended to the response. helper_spy = mocker.patch.object( splunk, "list_modified_investigations", return_value=[{"investigation_guid": "g-1", "update_time": 1737547610.56}], ) results_spy = mocker.patch.object(demisto, "results") splunk.get_modified_remote_data_command(**kwargs) assert helper_spy.call_count == 1 emitted_entries = results_spy.call_args[0][0] mirror_ids = {entry["EntryContext"]["mirrorRemoteId"] for entry in emitted_entries} assert {"F-1", "F-2", "g-1"}.issubset(mirror_ids) def test_given_investigations_not_selected_when_command_runs_then_helper_not_called(self, mocker): """ Given: - `fetch_event_types` is "Finding" only. When: - get_modified_remote_data_command runs. Then: - list_modified_investigations is NOT invoked, preserving BC for instances that have not opted into Investigation mirror-in. """ kwargs = self._build_kwargs(mocker) mocker.patch.object(demisto, "params", return_value={"timezone": "0", "fetch_event_types": "Finding"}) mocker.patch("SplunkPyV2.results.JSONResultsReader", return_value=[]) mocker.patch("SplunkPyV2.get_current_splunk_time", return_value="2021-02-09T17:41:30.589575+02:00") helper_spy = mocker.patch.object(splunk, "list_modified_investigations") mocker.patch.object(demisto, "results") splunk.get_modified_remote_data_command(**kwargs) assert helper_spy.call_count == 0 # ============================================================================================ # CLOSE-ON-MIRROR-IN TESTS FOR INVESTIGATIONS # ============================================================================================ class TestHandleClosedEntitiesForInvestigations: """Given investigation rows normalised by `parse_investigation` to expose the canonical closure keys (`status_label`, `status_end`), when the SAME `handle_closed_entities` helper that handles Findings is invoked, then investigation rows whose Splunk-side status indicates closure produce a `dbotIncidentClose` entry — and rows that are still open do not.""" def test_given_closed_investigation_when_helper_runs_then_close_entry_appended(self): """ Given: - An investigation row with `status_label="Closed"` (the canonical key Findings also use; populated for investigations by `parse_investigation` from the `status_name` field). When: - `handle_closed_entities` is called with `close_incident=True` equivalent (i.e. the caller already gated on the param). Then: - Exactly one `dbotIncidentClose` entry is appended to `entries`, keyed by the investigation guid via `mirrorRemoteId`, mirroring the Findings closure shape verbatim. """ entries: list[dict] = [] guid = "inv-guid-1" investigations_map = {guid: {"status_label": "Closed", "status_end": "false"}} splunk.handle_closed_entities( modified_entities_map=investigations_map, close_extra_labels=[], close_end_statuses=False, entries=entries, ) assert len(entries) == 1 entry = entries[0] assert entry["EntryContext"]["mirrorRemoteId"] == guid assert entry["Type"] == EntryType.NOTE assert entry["Contents"]["dbotIncidentClose"] is True assert "Closed" in entry["Contents"]["closeReason"] def test_given_open_investigation_when_helper_runs_then_no_close_entry_appended(self): """ Given: - An investigation row whose `status_label` is "New" (not closed, not in the configured close_extra_labels). When: - `handle_closed_entities` is called. Then: - `entries` remains empty — the helper must not fabricate closure for non-closed investigations. """ entries: list[dict] = [] investigations_map = {"inv-guid-2": {"status_label": "New", "status_end": "false"}} splunk.handle_closed_entities( modified_entities_map=investigations_map, close_extra_labels=["Custom"], close_end_statuses=True, entries=entries, ) assert entries == [] def test_given_close_incident_true_when_command_runs_then_handle_closed_entities_invoked_for_investigations(mocker): """ Given: - `fetch_event_types` includes "Investigation" and `close_incident=True`. - `list_modified_investigations` returns one row keyed by `investigation_guid`. When: - `get_modified_remote_data_command` runs. Then: - `handle_closed_entities` is invoked for the investigations map (in addition to / independently of the Findings call), proving the same mechanism is wired up for both event types. """ service = mocker.patch.object(client, "Service") mocker.patch.object(demisto, "params", return_value={"timezone": "0", "fetch_event_types": "Investigation"}) mocker.patch("SplunkPyV2.results.JSONResultsReader", return_value=[]) mocker.patch("SplunkPyV2.get_current_splunk_time", return_value="2021-02-09T17:41:30.589575+02:00") mocker.patch.object( splunk, "list_modified_investigations", return_value=[{"investigation_guid": "g-99", "update_time": 1737547610.56, "status_label": "Closed"}], ) mocker.patch.object(splunk, "enrich_with_splunk_notes_v2", return_value=[]) mocker.patch.object(demisto, "results") handle_spy = mocker.patch.object(splunk, "handle_closed_entities") splunk.get_modified_remote_data_command( service=service, args={"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, close_incident=True, close_end_statuses=True, close_extra_labels=["Custom"], mapper=splunk.UserMappingObject(service, False), ) # Assert handle_closed_entities was called with the investigations map (keyed # by investigation_guid). The Findings branch is skipped because # `fetch_event_types` is "Investigation"-only, so the only call must be the # investigation one. investigation_calls = [ call for call in handle_spy.call_args_list if "g-99" in (call.args[0] if call.args else call.kwargs.get("modified_entities_map", {})) ] assert len(investigation_calls) == 1 # ===================== Tests for splunk_update_investigation_command ===================== def _make_update_investigation_args(**overrides): """Helper that builds a baseline args dict for splunk-update-investigation tests.""" base = {"event_ids": "INV-1"} base.update(overrides) return base def test_splunk_update_investigation_command_name_only(mocker): """ Given: - args containing event_ids and only the new `name` field. When: - splunk_update_investigation_command is called. Then: - update_investigation_or_finding is called once with name=. - add_findings_to_investigation and add_investigation_note are NOT called. - The readable_output mirrors the splunk-finding-event-edit success format ("Splunk ES events updated successfully:" + per-id success line). """ args = _make_update_investigation_args(name="My new investigation name") mock_update = mocker.patch.object(splunk, "update_investigation_or_finding") mock_add_note = mocker.patch.object(splunk, "add_investigation_note") mock_add_findings = mocker.patch.object(splunk, "add_findings_to_investigation") result = splunk.splunk_update_investigation_command(service=MagicMock(), args=args) mock_update.assert_called_once_with( service=mocker.ANY, investigation_or_finding_id="INV-1", name="My new investigation name", ) mock_add_note.assert_not_called() mock_add_findings.assert_not_called() assert result.readable_output == ("Splunk ES events updated successfully:\nSuccessfully updated Splunk ES event INV-1") def test_splunk_update_investigation_command_description_only(mocker): """ Given: - args containing event_ids and only the new `description` field. When: - splunk_update_investigation_command is called. Then: - update_investigation_or_finding is called once with description=. - The readable_output uses the standard success format with the investigation id. """ args = _make_update_investigation_args(description="Updated description text") mock_update = mocker.patch.object(splunk, "update_investigation_or_finding") mocker.patch.object(splunk, "add_investigation_note") mocker.patch.object(splunk, "add_findings_to_investigation") result = splunk.splunk_update_investigation_command(service=MagicMock(), args=args) mock_update.assert_called_once_with( service=mocker.ANY, investigation_or_finding_id="INV-1", description="Updated description text", ) assert "Successfully updated Splunk ES event INV-1" in result.readable_output def test_splunk_update_investigation_command_name_and_description(mocker): """ Given: - args containing event_ids and both `name` and `description`. When: - splunk_update_investigation_command is called. Then: - update_investigation_or_finding is called once with both fields together. - The readable_output uses the standard success format with the investigation id. """ args = _make_update_investigation_args(name="New Name", description="New Description") mock_update = mocker.patch.object(splunk, "update_investigation_or_finding") mocker.patch.object(splunk, "add_investigation_note") mocker.patch.object(splunk, "add_findings_to_investigation") result = splunk.splunk_update_investigation_command(service=MagicMock(), args=args) mock_update.assert_called_once_with( service=mocker.ANY, investigation_or_finding_id="INV-1", name="New Name", description="New Description", ) assert "Successfully updated Splunk ES event INV-1" in result.readable_output def test_splunk_update_investigation_command_single_finding(mocker): """ Given: - args containing event_ids and a single `findings` value. When: - splunk_update_investigation_command is called. Then: - add_findings_to_investigation is called with a 1-element list. - update_investigation_or_finding is NOT called (no other update fields). - The readable_output uses the standard success format with the investigation id. """ args = _make_update_investigation_args(findings="FND-1") mock_update = mocker.patch.object(splunk, "update_investigation_or_finding") mock_add_findings = mocker.patch.object(splunk, "add_findings_to_investigation") result = splunk.splunk_update_investigation_command(service=MagicMock(), args=args) mock_update.assert_not_called() mock_add_findings.assert_called_once_with( service=mocker.ANY, investigation_id="INV-1", finding_ids=["FND-1"], ) assert "Successfully updated Splunk ES event INV-1" in result.readable_output def test_splunk_update_investigation_command_multiple_findings_csv(mocker): """ Given: - args containing event_ids and a CSV `findings` value (multiple ids). When: - splunk_update_investigation_command is called. Then: - add_findings_to_investigation is called once with a list of all parsed finding ids (single batch call, matching the Splunk API contract). - The readable_output uses the standard success format with the investigation id. """ args = _make_update_investigation_args(findings="FND-1,FND-2,FND-3") mocker.patch.object(splunk, "update_investigation_or_finding") mock_add_findings = mocker.patch.object(splunk, "add_findings_to_investigation") result = splunk.splunk_update_investigation_command(service=MagicMock(), args=args) mock_add_findings.assert_called_once_with( service=mocker.ANY, investigation_id="INV-1", finding_ids=["FND-1", "FND-2", "FND-3"], ) assert "Successfully updated Splunk ES event INV-1" in result.readable_output def test_splunk_update_investigation_command_combined_name_description_findings(mocker): """ Given: - args containing event_ids, name, description and findings together. When: - splunk_update_investigation_command is called. Then: - update_investigation_or_finding is called with name + description. - add_findings_to_investigation is called once with both finding ids. - The readable_output uses the standard success format with the investigation id. """ args = _make_update_investigation_args( name="Combined name", description="Combined description", findings="FND-1,FND-2", ) mock_update = mocker.patch.object(splunk, "update_investigation_or_finding") mock_add_findings = mocker.patch.object(splunk, "add_findings_to_investigation") result = splunk.splunk_update_investigation_command(service=MagicMock(), args=args) mock_update.assert_called_once_with( service=mocker.ANY, investigation_or_finding_id="INV-1", name="Combined name", description="Combined description", ) mock_add_findings.assert_called_once_with( service=mocker.ANY, investigation_id="INV-1", finding_ids=["FND-1", "FND-2"], ) assert "Successfully updated Splunk ES event INV-1" in result.readable_output def test_splunk_update_investigation_command_no_updatable_args_raises(mocker): """ Given: - args containing only event_ids and no updatable fields. When: - splunk_update_investigation_command is called. Then: - A DemistoException is raised with a clear message naming the supported fields. - No HTTP-related helper is called. """ args = {"event_ids": "INV-1"} mock_update = mocker.patch.object(splunk, "update_investigation_or_finding") mock_add_note = mocker.patch.object(splunk, "add_investigation_note") mock_add_findings = mocker.patch.object(splunk, "add_findings_to_investigation") with pytest.raises(DemistoException, match="At least one of"): splunk.splunk_update_investigation_command(service=MagicMock(), args=args) mock_update.assert_not_called() mock_add_note.assert_not_called() mock_add_findings.assert_not_called() def test_add_findings_to_investigation_posts_batch(): """ Given: - A list of two finding ids to append to an investigation. When: - add_findings_to_investigation is called. Then: - service.post is invoked once against the /findings endpoint with the JSON body containing the `finding_ids` array (single batch, matching the Splunk API). """ mock_service = MagicMock() mock_response = MagicMock() mock_response.body.read.return_value = b"{}" mock_service.post.return_value = mock_response splunk.add_findings_to_investigation( service=mock_service, investigation_id="INV-9", finding_ids=["FND-A", "FND-B"], ) assert mock_service.post.call_count == 1 args, kwargs = mock_service.post.call_args assert args[0] == "public/v2/investigations/INV-9/findings" body = json.loads(kwargs["body"]) assert body == {"finding_ids": ["FND-A", "FND-B"]} def test_splunk_update_investigation_command_findings_with_finding_times(mocker): """ Given: - args containing event_ids, multiple `findings` and a matching number of `finding_times`. When: - splunk_update_investigation_command is called. Then: - add_findings_to_investigation is called once with finding_ids and finding_times in the same order. - The underlying request body (via the helper) carries both arrays in parallel. - The readable_output uses the standard success format with the investigation id. """ args = _make_update_investigation_args( findings="FND-1,FND-2,FND-3", finding_times="1700000001,1700000002,1700000003", ) mock_add_findings = mocker.patch.object(splunk, "add_findings_to_investigation") result = splunk.splunk_update_investigation_command(service=MagicMock(), args=args) mock_add_findings.assert_called_once_with( service=mocker.ANY, investigation_id="INV-1", finding_ids=["FND-1", "FND-2", "FND-3"], finding_times=["1700000001", "1700000002", "1700000003"], ) assert "Successfully updated Splunk ES event INV-1" in result.readable_output def test_splunk_update_investigation_command_finding_times_without_findings_raises(mocker): """ Given: - args containing event_ids and `finding_times` but no `findings`. When: - splunk_update_investigation_command is called. Then: - A DemistoException is raised with a clear message stating finding_times requires findings. - add_findings_to_investigation is NOT called. """ args = _make_update_investigation_args(finding_times="1700000001,1700000002") mock_add_findings = mocker.patch.object(splunk, "add_findings_to_investigation") with pytest.raises(DemistoException, match="'finding_times' was provided without 'findings'"): splunk.splunk_update_investigation_command(service=MagicMock(), args=args) mock_add_findings.assert_not_called() def test_splunk_update_investigation_command_findings_with_multiple_event_ids_raises(mocker): """ Given: - args containing multiple event_ids (CSV) together with a `findings` value. When: - splunk_update_investigation_command is called. Then: - A DemistoException is raised stating findings can only be used with a single investigation. - add_findings_to_investigation is NOT called (guard fires before any HTTP call). """ args = _make_update_investigation_args(event_ids="ES-1,ES-2", findings="FND-1") mock_add_findings = mocker.patch.object(splunk, "add_findings_to_investigation") mocker.patch.object(splunk, "update_investigation_or_finding") mocker.patch.object(splunk, "add_investigation_note") with pytest.raises( DemistoException, match="'findings' \\(and 'finding_times'\\) can only be used when updating a single investigation", ): splunk.splunk_update_investigation_command(service=MagicMock(), args=args) mock_add_findings.assert_not_called() def test_splunk_update_investigation_command_finding_times_with_multiple_event_ids_raises(mocker): """ Given: - args containing multiple event_ids (CSV) together with `findings` and `finding_times`. When: - splunk_update_investigation_command is called. Then: - A DemistoException is raised stating findings/finding_times require a single investigation. - add_findings_to_investigation is NOT called (guard fires before any HTTP call). """ args = _make_update_investigation_args( event_ids="ES-1,ES-2", findings="FND-1", finding_times="1700000001", ) mock_add_findings = mocker.patch.object(splunk, "add_findings_to_investigation") mocker.patch.object(splunk, "update_investigation_or_finding") mocker.patch.object(splunk, "add_investigation_note") with pytest.raises( DemistoException, match="'findings' \\(and 'finding_times'\\) can only be used when updating a single investigation", ): splunk.splunk_update_investigation_command(service=MagicMock(), args=args) mock_add_findings.assert_not_called() def test_splunk_update_investigation_command_multiple_event_ids_without_findings_still_works(mocker): """ Given: - args containing multiple event_ids (CSV) and `name` but NO findings/finding_times. When: - splunk_update_investigation_command is called. Then: - No exception is raised; the existing multi-investigation update path is preserved. - update_investigation_or_finding is called once per investigation id with name=. - add_findings_to_investigation is NOT called. """ args = _make_update_investigation_args(event_ids="ES-1,ES-2", name="updated") mock_update = mocker.patch.object(splunk, "update_investigation_or_finding") mock_add_note = mocker.patch.object(splunk, "add_investigation_note") mock_add_findings = mocker.patch.object(splunk, "add_findings_to_investigation") result = splunk.splunk_update_investigation_command(service=MagicMock(), args=args) assert mock_update.call_count == 2 mock_update.assert_any_call( service=mocker.ANY, investigation_or_finding_id="ES-1", name="updated", ) mock_update.assert_any_call( service=mocker.ANY, investigation_or_finding_id="ES-2", name="updated", ) mock_add_note.assert_not_called() mock_add_findings.assert_not_called() assert "Successfully updated Splunk ES event ES-1" in result.readable_output assert "Successfully updated Splunk ES event ES-2" in result.readable_output # --------------------------------------------------------------------------- # # Tests for splunk-investigation-create # # --------------------------------------------------------------------------- # def test_splunk_create_investigation_command_minimal_args(mocker): """ Given: - Args containing only the required `name` argument. When: - splunk_create_investigation_command is called. Then: - _es_rest_request is called once with POST and a payload containing only `name`. - CommandResults outputs_prefix is `Splunk.CreateInvestigation` with the returned guid. - The readable output mentions the investigation GUID. """ expected_guid = "11111111-2222-3333-4444-555555555555" mock_es_request = mocker.patch.object( splunk, "_es_rest_request", return_value={"investigation_guid": expected_guid}, ) args = {"name": "My Investigation"} result = splunk.splunk_create_investigation_command(service=MagicMock(), args=args) mock_es_request.assert_called_once() _, called_kwargs = mock_es_request.call_args called_args_positional = mock_es_request.call_args.args # signature: _es_rest_request(service, method, path, body=..., query=...) assert called_args_positional[1] == "POST" assert called_args_positional[2] == "public/v2/investigations" assert called_kwargs["body"] == {"name": "My Investigation"} assert result.outputs_prefix == "Splunk.Investigation" assert result.outputs_key_field == "investigation_guid" assert result.outputs == {"investigation_guid": expected_guid} assert "Investigation created successfully" in result.readable_output assert expected_guid in result.readable_output def test_splunk_create_investigation_command_all_fields_forwarded(mocker): """ Given: - Args containing all supported create fields and an extra unsupported field. When: - splunk_create_investigation_command is called. Then: - Only fields in INVESTIGATION_CREATE_FIELDS are forwarded in the payload. - The unsupported `extra_field` argument is ignored. """ mock_es_request = mocker.patch.object( splunk, "_es_rest_request", return_value={"investigation_guid": "guid-1"}, ) args = { "name": "Inv-Full", "description": "desc", "investigation_type": "default", "status": "New", "disposition": "Undetermined", "owner": "admin", "urgency": "high", "sensitivity": "Amber", "extra_field": "should-be-ignored", } splunk.splunk_create_investigation_command(service=MagicMock(), args=args) body = mock_es_request.call_args.kwargs["body"] assert body == { "name": "Inv-Full", "description": "desc", "investigation_type": "default", "status": "New", "disposition": "Undetermined", "owner": "admin", "urgency": "high", "sensitivity": "Amber", } assert "extra_field" not in body def test_splunk_create_investigation_command_missing_name_raises(mocker): """ Given: - Args missing the required `name` argument. When: - splunk_create_investigation_command is called. Then: - A DemistoException is raised stating that `name` is required. - No HTTP request is performed. """ mock_es_request = mocker.patch.object(splunk, "_es_rest_request") with pytest.raises(DemistoException, match="`name` is a required argument"): splunk.splunk_create_investigation_command(service=MagicMock(), args={"description": "no name"}) mock_es_request.assert_not_called() def test_splunk_create_investigation_command_non_dict_response(mocker): """ Given: - _es_rest_request returns a non-dict response (e.g., a list). When: - splunk_create_investigation_command is called. Then: - The command does not raise. - outputs is None (no investigation_guid available). - The raw_response still preserves the original response. """ mocker.patch.object(splunk, "_es_rest_request", return_value=["unexpected"]) result = splunk.splunk_create_investigation_command(service=MagicMock(), args={"name": "X"}) assert result.outputs is None assert result.raw_response == ["unexpected"] assert "Investigation created successfully" in result.readable_output # --------------------------------------------------------------------------- # # Tests for splunk-investigation-list # # --------------------------------------------------------------------------- # def test_splunk_list_investigations_command_no_args_returns_list(mocker): """ Given: - No filter args provided. - _es_rest_request returns a list of two investigation dicts. When: - splunk_list_investigations_command is called. Then: - GET public/v2/investigations is called with an empty query. - outputs is the full list (length 2). - The HR title reflects the count. """ investigations = [ { "investigation_guid": "guid-1", "investigation_id": "ES-00001", "name": "Inv 1", "status_name": "New", "owner": "admin", }, { "investigation_guid": "guid-2", "investigation_id": "ES-00002", "name": "Inv 2", "status_name": "In progress", "owner": "analyst", }, ] mock_es_request = mocker.patch.object( splunk, "_es_rest_request", return_value=investigations, ) result = splunk.splunk_list_investigations_command(service=MagicMock(), args={}) called_args = mock_es_request.call_args.args called_kwargs = mock_es_request.call_args.kwargs assert called_args[1] == "GET" assert called_args[2] == "public/v2/investigations" assert called_kwargs["query"] == {} assert result.outputs_prefix == "Splunk.Investigation" assert result.outputs_key_field == "investigation_guid" assert isinstance(result.outputs, list) assert len(result.outputs) == 2 assert "2 Investigations Found" in result.readable_output def test_splunk_list_investigations_command_single_result_returns_dict(mocker): """ Given: - _es_rest_request returns a list with a single investigation dict. When: - splunk_list_investigations_command is called. Then: - outputs is the single investigation dict (not wrapped in a list). - HR title uses the singular form "Investigation". """ investigation = { "investigation_guid": "guid-1", "investigation_id": "ES-00001", "name": "Solo", "status_name": "New", } mocker.patch.object(splunk, "_es_rest_request", return_value=[investigation]) result = splunk.splunk_list_investigations_command(service=MagicMock(), args={}) assert isinstance(result.outputs, dict) assert result.outputs == investigation assert "1 Investigation Found" in result.readable_output def test_splunk_list_investigations_command_with_filters_forwarded_as_csv(mocker): """ Given: - Args with multi-value filters (`investigation_ids`, `status`, `urgency`) and scalar filters (`limit`, `offset`, `sort`). When: - splunk_list_investigations_command is called. Then: - The XSOAR-facing `investigation_ids` arg is mapped to the Splunk API `ids` query parameter and forwarded as a CSV string. - Other multi-value args are forwarded as CSV strings. - Scalar args are forwarded unchanged. - The HR title includes the provided ids list. """ mock_es_request = mocker.patch.object( splunk, "_es_rest_request", return_value=[{"investigation_guid": "guid-1", "investigation_id": "ES-00001", "name": "n"}], ) args = { "investigation_ids": "ES-00001,ES-00002", "status": "New,In progress", "urgency": "high,critical", "limit": "10", "offset": "5", "sort": "create_time:desc", } result = splunk.splunk_list_investigations_command(service=MagicMock(), args=args) query = mock_es_request.call_args.kwargs["query"] assert query["ids"] == "ES-00001,ES-00002" assert query["status"] == "New,In progress" assert query["urgency"] == "high,critical" assert query["limit"] == "10" assert query["offset"] == "5" assert query["sort"] == "create_time:desc" assert "ES-00001, ES-00002" in result.readable_output def test_splunk_list_investigations_command_empty_response(mocker): """ Given: - _es_rest_request returns an empty list. When: - splunk_list_investigations_command is called. Then: - CommandResults has a friendly "No investigations found" message. - No outputs / outputs_prefix is set. """ mocker.patch.object(splunk, "_es_rest_request", return_value=[]) result = splunk.splunk_list_investigations_command(service=MagicMock(), args={}) assert result.outputs is None assert result.outputs_prefix is None assert "No investigations found" in result.readable_output def test_splunk_list_investigations_command_dict_response_with_nested_list(mocker): """ Given: - _es_rest_request returns a dict whose value is the list of investigations (e.g., {"results": [...]}). When: - splunk_list_investigations_command is called. Then: - The nested list is correctly unwrapped into outputs. """ investigations = [ {"investigation_guid": "guid-1", "investigation_id": "ES-00001", "name": "A"}, {"investigation_guid": "guid-2", "investigation_id": "ES-00002", "name": "B"}, ] mocker.patch.object(splunk, "_es_rest_request", return_value={"results": investigations}) result = splunk.splunk_list_investigations_command(service=MagicMock(), args={}) assert isinstance(result.outputs, list) assert len(result.outputs) == 2 assert result.outputs[0]["investigation_guid"] == "guid-1" def test_splunk_list_investigations_command_dict_response_single_investigation(mocker): """ Given: - _es_rest_request returns a single investigation dict (not wrapped in a list). When: - splunk_list_investigations_command is called. Then: - The single dict is treated as a list of one investigation. - outputs is the investigation dict. """ investigation = { "investigation_guid": "guid-single", "investigation_id": "ES-00009", "name": "Single", } mocker.patch.object(splunk, "_es_rest_request", return_value=investigation) result = splunk.splunk_list_investigations_command(service=MagicMock(), args={}) assert result.outputs == investigation assert "1 Investigation Found" in result.readable_output # --------------------------------------------------------------------------- # Configuration (.conf) file / stanza commands # --------------------------------------------------------------------------- def _mock_stanza(name, content=None, access=None): """Builds a mock Splunk SDK Stanza object with the given name, content and access.""" stanza = MagicMock() stanza.name = name stanza.content = content or {} stanza.access = access or {} return stanza def _mock_conf_file(name, stanzas=None): """Builds a mock Splunk SDK ConfigurationFile that behaves like a dict/collection of stanzas.""" stanzas = stanzas or {} conf_file = MagicMock() conf_file.name = name conf_file.__contains__.side_effect = lambda key: key in stanzas conf_file.__getitem__.side_effect = lambda key: stanzas[key] conf_file.list.return_value = list(stanzas.values()) return conf_file def _mock_service_with_confs(conf_files): """Builds a mock service whose service.confs behaves like a dict/collection of conf files.""" service = MagicMock() service.confs.__contains__.side_effect = lambda key: key in conf_files service.confs.__getitem__.side_effect = lambda key: conf_files[key] if key in conf_files else _raise_key_error(key) service.confs.list.return_value = list(conf_files.values()) return service def _raise_key_error(key): raise KeyError(key) def test_parse_key_value_pairs_valid(): """ Given: A valid JSON object string. When: parse_key_value_pairs is called. Then: The parsed dict is returned. """ assert splunk.parse_key_value_pairs('{"a": "1", "b": "2"}') == {"a": "1", "b": "2"} def test_parse_key_value_pairs_empty(): """ Given: An empty/None key_value_pairs argument. When: parse_key_value_pairs is called. Then: An empty dict is returned. """ assert splunk.parse_key_value_pairs(None) == {} assert splunk.parse_key_value_pairs("") == {} def test_parse_key_value_pairs_invalid_json(): """ Given: A non-JSON string. When: parse_key_value_pairs is called. Then: A DemistoException is raised. """ with pytest.raises(DemistoException, match="valid JSON object string"): splunk.parse_key_value_pairs("not-json") def test_parse_key_value_pairs_not_object(): """ Given: A JSON array (not an object). When: parse_key_value_pairs is called. Then: A DemistoException is raised. """ with pytest.raises(DemistoException, match="must be a JSON object"): splunk.parse_key_value_pairs('["a", "b"]') def test_splunk_configuration_file_list(): """ Given: A service with two configuration files. When: splunk_configuration_file_list is called. Then: Both files are returned under Splunk.ConfigurationFile with the app name. """ conf_files = {"transforms": _mock_conf_file("transforms"), "props": _mock_conf_file("props")} service = _mock_service_with_confs(conf_files) result = splunk.splunk_configuration_file_list(service, {"app": "search"}) assert result.outputs_prefix == "Splunk.ConfigurationFile" assert {c["FileName"] for c in result.outputs} == {"transforms", "props"} assert all(c["App"] == "search" for c in result.outputs) def test_splunk_configuration_file_create(): """ Given: A service and a new conf file name. When: splunk_configuration_file_create is called. Then: service.confs.create is called and a success message is returned. """ service = MagicMock() result = splunk.splunk_configuration_file_create(service, {"conf_file_name": "my_conf", "app": "search"}) service.confs.create.assert_called_once_with("my_conf") assert "was created successfully" in result.readable_output def test_splunk_configuration_stanza_create_with_attributes(): """ Given: A service with a conf file and key_value_pairs. When: splunk_configuration_stanza_create is called. Then: The stanza is created and its attributes submitted. """ created_stanza = MagicMock() conf_file = _mock_conf_file("transforms") conf_file.create.return_value = created_stanza service = _mock_service_with_confs({"transforms": conf_file}) result = splunk.splunk_configuration_stanza_create( service, {"conf_file": "transforms", "stanza_name": "my_stanza", "key_value_pairs": '{"external_type": "kvstore"}'} ) conf_file.create.assert_called_once_with("my_stanza") created_stanza.submit.assert_called_once_with({"external_type": "kvstore"}) assert "was created successfully" in result.readable_output def test_splunk_configuration_stanza_create_missing_conf_file(): """ Given: A service that does not contain the requested conf file. When: splunk_configuration_stanza_create is called. Then: A DemistoException is raised about the missing configuration file. """ service = _mock_service_with_confs({}) with pytest.raises(DemistoException, match="Configuration file 'transforms' was not found"): splunk.splunk_configuration_stanza_create(service, {"conf_file": "transforms", "stanza_name": "my_stanza"}) def test_splunk_configuration_stanza_list_all(): """ Given: A conf file with two stanzas. When: splunk_configuration_stanza_list is called without stanza_name. Then: All stanza names are returned under Splunk.ConfigurationStanza. """ stanzas = { "s1": _mock_stanza("s1", access={"app": "search", "owner": "nobody", "sharing": "app"}), "s2": _mock_stanza("s2", access={"app": "search", "owner": "nobody", "sharing": "app"}), } conf_file = _mock_conf_file("transforms", stanzas) service = _mock_service_with_confs({"transforms": conf_file}) result = splunk.splunk_configuration_stanza_list(service, {"conf_file": "transforms"}) assert result.outputs_prefix == "Splunk.ConfigurationStanza" assert {s["StanzaName"] for s in result.outputs} == {"s1", "s2"} def test_splunk_configuration_stanza_list_single(): """ Given: A conf file with a specific stanza that has content. When: splunk_configuration_stanza_list is called with that stanza_name. Then: The stanza content is returned. """ stanzas = {"s1": _mock_stanza("s1", content={"collection": "my_col"}, access={"app": "search"})} conf_file = _mock_conf_file("transforms", stanzas) service = _mock_service_with_confs({"transforms": conf_file}) result = splunk.splunk_configuration_stanza_list(service, {"conf_file": "transforms", "stanza_name": "s1"}) assert result.outputs["StanzaName"] == "s1" assert result.outputs["Content"] == {"collection": "my_col"} def test_splunk_configuration_stanza_list_single_not_found(): """ Given: A conf file that does not contain the requested stanza. When: splunk_configuration_stanza_list is called with that stanza_name. Then: A DemistoException is raised. """ conf_file = _mock_conf_file("transforms", {}) service = _mock_service_with_confs({"transforms": conf_file}) with pytest.raises(DemistoException, match="Stanza 'missing' was not found"): splunk.splunk_configuration_stanza_list(service, {"conf_file": "transforms", "stanza_name": "missing"}) def test_splunk_configuration_stanza_update(): """ Given: A conf file with an existing stanza and key_value_pairs. When: splunk_configuration_stanza_update is called. Then: The stanza's submit is called with the parsed attributes. """ stanza = _mock_stanza("s1") conf_file = _mock_conf_file("transforms", {"s1": stanza}) service = _mock_service_with_confs({"transforms": conf_file}) result = splunk.splunk_configuration_stanza_update( service, {"conf_file": "transforms", "stanza_name": "s1", "key_value_pairs": '{"attr": "new"}'} ) stanza.submit.assert_called_once_with({"attr": "new"}) assert "was updated successfully" in result.readable_output def test_splunk_configuration_stanza_update_requires_key_value_pairs(): """ Given: An update request without key_value_pairs. When: splunk_configuration_stanza_update is called. Then: A DemistoException is raised. """ conf_file = _mock_conf_file("transforms", {"s1": _mock_stanza("s1")}) service = _mock_service_with_confs({"transforms": conf_file}) with pytest.raises(DemistoException, match='"key_value_pairs" argument is required'): splunk.splunk_configuration_stanza_update(service, {"conf_file": "transforms", "stanza_name": "s1"}) def test_splunk_configuration_stanza_update_stanza_not_found(): """ Given: A conf file that does not contain the stanza to update. When: splunk_configuration_stanza_update is called. Then: A DemistoException is raised. """ conf_file = _mock_conf_file("transforms", {}) service = _mock_service_with_confs({"transforms": conf_file}) with pytest.raises(DemistoException, match="Stanza 's1' was not found"): splunk.splunk_configuration_stanza_update( service, {"conf_file": "transforms", "stanza_name": "s1", "key_value_pairs": '{"a": "b"}'} ) def test_splunk_configuration_stanza_delete(): """ Given: A conf file that contains the stanza to delete. When: splunk_configuration_stanza_delete is called. Then: conf_file.delete is called with the stanza name and a success message is returned. """ conf_file = _mock_conf_file("transforms", {"s1": _mock_stanza("s1")}) service = _mock_service_with_confs({"transforms": conf_file}) result = splunk.splunk_configuration_stanza_delete(service, {"conf_file": "transforms", "stanza_name": "s1"}) conf_file.delete.assert_called_once_with("s1") assert "was deleted successfully" in result.readable_output def test_splunk_configuration_stanza_delete_stanza_not_found(): """ Given: A conf file that does not contain the stanza to delete. When: splunk_configuration_stanza_delete is called. Then: A DemistoException is raised and delete is not called. """ conf_file = _mock_conf_file("transforms", {}) service = _mock_service_with_confs({"transforms": conf_file}) with pytest.raises(DemistoException, match="Stanza 's1' was not found"): splunk.splunk_configuration_stanza_delete(service, {"conf_file": "transforms", "stanza_name": "s1"}) conf_file.delete.assert_not_called() def test_splunk_configuration_stanza_delete_conf_file_not_found(): """ Given: A service that does not contain the requested conf file. When: splunk_configuration_stanza_delete is called. Then: A DemistoException is raised about the missing configuration file. """ service = _mock_service_with_confs({}) with pytest.raises(DemistoException, match="Configuration file 'transforms' was not found"): splunk.splunk_configuration_stanza_delete(service, {"conf_file": "transforms", "stanza_name": "s1"}) # =========== Enterprise Security version / mirror-out finding_time helpers =========== def _mock_service_with_es_app(version): """Build a mock Splunk service whose ES app returns the given version.""" es_app = MagicMock() es_app.content = {"version": version} service = MagicMock() service.apps = {splunk.ES_APP_NAME: es_app} return service def test_get_enterprise_security_version_returns_version(): """ Given: A Splunk service whose Enterprise Security app reports version "8.2.0". When: get_enterprise_security_version is called. Then: The reported version string "8.2.0" is returned. """ service = _mock_service_with_es_app("8.2.0") result = splunk.get_enterprise_security_version(service) assert result == "8.2.0" def test_get_enterprise_security_version_missing_version_key(): """ Given: A Splunk service whose Enterprise Security app content has no "version" key. When: get_enterprise_security_version is called. Then: "unknown" is returned. """ es_app = MagicMock() es_app.content = {} service = MagicMock() service.apps = {splunk.ES_APP_NAME: es_app} result = splunk.get_enterprise_security_version(service) assert result == "unknown" def test_get_enterprise_security_version_app_not_installed(): """ Given: A Splunk service whose apps lookup raises (ES app not installed). When: get_enterprise_security_version is called. Then: The exception is swallowed and "unknown" is returned. """ service = MagicMock() service.apps.__getitem__.side_effect = KeyError("SplunkEnterpriseSecuritySuite") result = splunk.get_enterprise_security_version(service) assert result == "unknown" @pytest.mark.parametrize( "version, target_version, expected", [ ("8.2", "8.2", True), ("8.2.0", "8.2", True), ("8.2.5", "8.2", True), ("8.1.9", "8.2", False), ("8.3.0", "8.2", False), ("9.0.0", "8.2", False), ("unknown", "8.2", False), ("", "8.2", False), ("8.3.1", "8.3", True), ("8.2.5", "8.3", False), ("8.2", "unknown", False), ], ) def test_is_es_version(version, target_version, expected): """ Given: An Enterprise Security version string and a target version. When: is_es_version is called. Then: True is returned only when the major/minor match; unparseable values return False. """ assert splunk.is_es_version(version, target_version) is expected def test_get_finding_time_for_es_notable_time_returns_time_on_8_2(mocker: MockerFixture): """ Given: An ES 8.2.x service and mirrored data containing a "notable_time" value. When: get_finding_time_for_es_notable_time is called. Then: The notable_time value is returned as a string. """ mocker.patch.object(splunk, "get_enterprise_security_version", return_value="8.2.0") service = MagicMock() result = splunk.get_finding_time_for_es_notable_time(service, {"notable_time": "2026-01-01T00:00:00.000Z"}) assert result == "2026-01-01T00:00:00.000Z" def test_get_finding_time_for_es_notable_time_none_when_not_8_2(mocker: MockerFixture): """ Given: An ES 8.3.x service (out of the 8.2.x range) and mirrored data with a notable_time. When: get_finding_time_for_es_notable_time is called. Then: None is returned because finding_time is only required on 8.2.x. """ mocker.patch.object(splunk, "get_enterprise_security_version", return_value="8.3.0") service = MagicMock() result = splunk.get_finding_time_for_es_notable_time(service, {"notable_time": "2026-01-01T00:00:00.000Z"}) assert result is None def test_get_finding_time_for_es_notable_time_none_when_no_time_in_data(mocker: MockerFixture): """ Given: An ES 8.2.x service but mirrored data that has no "notable_time" field. When: get_finding_time_for_es_notable_time is called. Then: None is returned since there is no finding time to send. """ mocker.patch.object(splunk, "get_enterprise_security_version", return_value="8.2.1") service = MagicMock() result = splunk.get_finding_time_for_es_notable_time(service, {"other": "value"}) assert result is None def test_get_finding_time_for_es_notable_time_none_when_data_is_none(mocker: MockerFixture): """ Given: An ES 8.2.x service and data is None. When: get_finding_time_for_es_notable_time is called. Then: None is returned without raising. """ mocker.patch.object(splunk, "get_enterprise_security_version", return_value="8.2.0") service = MagicMock() result = splunk.get_finding_time_for_es_notable_time(service, None) assert result is None