import pytest from datetime import datetime from CommonServerPython import DemistoException from SearchCases import prepare_start_end_time, main, extract_ids, add_cases_extra_data, get_case_extra_data def test_prepare_start_end_time_normal(monkeypatch): """ GIVEN valid start_time and end_time arguments in ISO format WHEN prepare_start_end_time is called THEN it sets gte_creation_time and lte_creation_time correctly in the args dict """ args = {"start_time": "2025-09-01T12:00:00", "end_time": "2025-09-02T13:00:00"} prepare_start_end_time(args) assert args["gte_creation_time"] == "2025-09-01T12:00:00" assert args["lte_creation_time"] == "2025-09-02T13:00:00" def test_prepare_start_end_time_end_without_start(): """ GIVEN only end_time is provided in args WHEN prepare_start_end_time is called THEN it raises DemistoException because start_time is required if end_time is provided """ args = {"end_time": "2025-09-02T13:00:00"} with pytest.raises(DemistoException): prepare_start_end_time(args) def test_prepare_start_end_time_only_start(monkeypatch): """ GIVEN only start_time is provided in args WHEN prepare_start_end_time is called THEN it sets gte_creation_time and lte_creation_time (lte_creation_time defaults to now) """ args = {"start_time": "2025-09-01T12:00:00"} monkeypatch.setattr("SearchCases.datetime", datetime) prepare_start_end_time(args) assert "gte_creation_time" in args assert "lte_creation_time" in args def test_prepare_start_end_time_both_empty(): """ GIVEN no start_time or end_time in args WHEN prepare_start_end_time is called THEN it does not set gte_creation_time or lte_creation_time """ args = {} prepare_start_end_time(args) assert "gte_creation_time" not in args assert "lte_creation_time" not in args def test_prepare_start_end_time_unparseable(): """ GIVEN start_time and end_time are unparseable strings WHEN prepare_start_end_time is called THEN it does not set gte_creation_time or lte_creation_time """ args = {"start_time": "not-a-date", "end_time": "also-not-a-date"} prepare_start_end_time(args) assert "gte_creation_time" not in args assert "lte_creation_time" not in args def test_prepare_start_end_time_only_end(): """ GIVEN only end_time is provided in args (again) WHEN prepare_start_end_time is called THEN it raises DemistoException because start_time is required if end_time is provided """ args = {"end_time": "2025-09-02T13:00:00"} try: prepare_start_end_time(args) except DemistoException as e: assert "start_time must be provided" in str(e) def test_prepare_start_end_time_relative(monkeypatch): """ GIVEN start_time and end_time as relative date strings WHEN prepare_start_end_time is called THEN it sets gte_creation_time and lte_creation_time in the args dict """ args = {"start_time": "1 day ago", "end_time": "now"} prepare_start_end_time(args) assert "gte_creation_time" in args assert "lte_creation_time" in args def test_main_success(mocker): """ GIVEN valid demisto.args and executeCommand returns a valid result WHEN main is called THEN return_results is called with the expected output """ mock_args = {"start_time": "2025-09-01T12:00:00", "end_time": "2025-09-02T13:00:00", "page_size": 5} mocker.patch("demistomock.args", return_value=mock_args.copy()) mocker.patch( "demistomock.executeCommand", return_value=[ { "EntryContext": {"Core.Case": [{"case_id": "1"}]}, "HumanReadable": "ok", "Type": 1, } ], ) mocker.patch("SearchCases.prepare_start_end_time") mocked_return_results = mocker.patch("SearchCases.return_results") main() mocked_return_results.assert_called() def test_main_error(mocker): """ GIVEN executeCommand returns an error result WHEN main is called THEN return_error is called """ mock_args = {"start_time": "2025-09-01T12:00:00", "end_time": "2025-09-02T13:00:00", "page_size": 5} mocker.patch("demistomock.args", return_value=mock_args.copy()) mocker.patch( "demistomock.executeCommand", return_value=[ { "Type": 4, "ContentsFormat": "text", "Contents": "error", "HumanReadable": "fail", "EntryContext": {}, "ModuleName": "", "Brand": "", "ID": "", "FileID": "", } ], ) mocker.patch("SearchCases.prepare_start_end_time") mocked_return_error = mocker.patch("SearchCases.return_error") mocker.patch("SearchCases.is_error", return_value=True) mocker.patch("SearchCases.get_error", return_value="fail") main() mocked_return_error.assert_called() def test_get_case_extra_data_command_error(mocker): """ GIVEN execute_command that raises an exception WHEN get_case_extra_data is called THEN it raises the exception """ mocker.patch("SearchCases.execute_command", side_effect=Exception("Command failed")) with pytest.raises(Exception): get_case_extra_data({"case_id": "123"}) def test_add_cases_extra_data_single_case(mocker): """ GIVEN a list with one case containing case_id WHEN add_cases_extra_data is called THEN it adds CaseExtraData to the case by calling get_case_extra_data """ mock_get_case_extra_data = mocker.patch("SearchCases.get_case_extra_data") mock_extra_data = {"issues": {"total_count": 5}, "alerts": {"total_count": 3}} mock_get_case_extra_data.return_value = mock_extra_data case_data = [{"case_id": "123", "case_name": "Test Case"}] result = add_cases_extra_data(case_data) mock_get_case_extra_data.assert_called_once_with({"case_id": "123", "limit": 1000}) assert result[0]["CaseExtraData"] == mock_extra_data assert result[0]["case_id"] == "123" assert result[0]["case_name"] == "Test Case" def test_add_cases_extra_data_multiple_cases(mocker): """ GIVEN a list with multiple cases containing case_ids WHEN add_cases_extra_data is called THEN it adds CaseExtraData to each case by calling get_case_extra_data for each """ mock_get_case_extra_data = mocker.patch("SearchCases.get_case_extra_data") mock_extra_data_1 = {"issues": {"total_count": 5}} mock_extra_data_2 = {"issues": {"total_count": 10}} mock_get_case_extra_data.side_effect = [mock_extra_data_1, mock_extra_data_2] case_data = [{"case_id": "123", "case_name": "Test Case 1"}, {"case_id": "456", "case_name": "Test Case 2"}] result = add_cases_extra_data(case_data) assert mock_get_case_extra_data.call_count == 2 mock_get_case_extra_data.assert_any_call({"case_id": "123", "limit": 1000}) mock_get_case_extra_data.assert_any_call({"case_id": "456", "limit": 1000}) assert result[0]["CaseExtraData"] == mock_extra_data_1 assert result[1]["CaseExtraData"] == mock_extra_data_2 def test_add_cases_extra_data_empty_list(mocker): """ GIVEN an empty list of cases WHEN add_cases_extra_data is called THEN it returns empty list without calling get_case_extra_data """ mock_get_case_extra_data = mocker.patch("SearchCases.get_case_extra_data") case_data = [] result = add_cases_extra_data(case_data) mock_get_case_extra_data.assert_not_called() assert result == [] def test_get_case_extra_data_normal(mocker): """ GIVEN valid args for get_case_extra_data WHEN get_case_extra_data is called THEN it returns properly formatted extra data with issue_ids, network_artifacts, and file_artifacts """ mock_case_extra_data = { "issues": {"data": [{"issue_id": "101"}, {"issue_id": "102"}]}, "network_artifacts": [{"ip": "1.2.3.4"}], "file_artifacts": [{"filename": "test.exe"}], } mocker.patch("SearchCases.execute_command", return_value=mock_case_extra_data) mocker.patch("SearchCases.extract_ids", return_value=["101", "102"]) mock_debug = mocker.patch("demistomock.debug") args = {"case_id": "123"} result = get_case_extra_data(args) assert result["issue_ids"] == ["101", "102"] assert result["network_artifacts"] == [{"ip": "1.2.3.4"}] assert result["file_artifacts"] == [{"filename": "test.exe"}] assert mock_debug.call_count == 2 def test_get_case_extra_data_no_artifacts(mocker): """ GIVEN case extra data with no network or file artifacts WHEN get_case_extra_data is called THEN it returns None for missing artifact types """ mock_case_extra_data = {"issues": {"data": [{"issue_id": "101"}]}} mocker.patch("SearchCases.execute_command", return_value=mock_case_extra_data) mocker.patch("SearchCases.extract_ids", return_value=["101"]) mocker.patch("demistomock.debug") args = {"case_id": "123"} result = get_case_extra_data(args) assert result["issue_ids"] == ["101"] assert result["network_artifacts"] is None assert result["file_artifacts"] is None def test_get_case_extra_data_empty_issues(mocker): """ GIVEN case extra data with empty issues WHEN get_case_extra_data is called THEN it returns empty issue_ids list """ mock_case_extra_data = {"issues": {"data": []}, "network_artifacts": [], "file_artifacts": []} mocker.patch("SearchCases.execute_command", return_value=mock_case_extra_data) mocker.patch("SearchCases.extract_ids", return_value=[]) mocker.patch("demistomock.debug") args = {"case_id": "123"} result = get_case_extra_data(args) assert result["issue_ids"] == [] assert result["network_artifacts"] == [] assert result["file_artifacts"] == [] def test_extract_ids_mixed_valid_invalid_items(): """ GIVEN case_extra_data with mix of valid items, items missing issue_id, and non-dict items WHEN extract_ids is called THEN it returns only valid issue_ids """ case_extra_data = { "issues": { "data": [ {"issue_id": "100", "name": "Issue 1"}, {"name": "Issue 2"}, "not_a_dict", {"issue_id": "101", "severity": "high"}, None, {"issue_id": "102"}, ] } } result = extract_ids(case_extra_data) assert result == ["100", "101", "102"] def test_extract_ids_non_list_data(mocker): """ GIVEN case_extra_data where issues.data is not a list WHEN extract_ids is called THEN it returns empty list """ case_extra_data = {"issues": {"data": {"not_a_list": True}}} result = extract_ids(case_extra_data) assert result == [] def test_extract_ids_no_issues(): """ GIVEN case_extra_data with no issues key WHEN extract_ids is called THEN it returns empty list """ case_extra_data = { "case": { "aggregated_score": None, "assigned_user_mail": None, "assigned_user_pretty_name": None, "case_domain": "DOMAIN_POSTURE", "case_id": "62", } } result = extract_ids(case_extra_data) assert result == [] def test_extract_ids_no_issues_data(): """ GIVEN case_extra_data with issues key but no data key WHEN extract_ids is called THEN it returns empty list """ case_extra_data = { "case": { "aggregated_score": None, "assigned_user_mail": None, "assigned_user_pretty_name": None, "case_domain": "DOMAIN_POSTURE", "case_id": "62", }, "issues": {}, } result = extract_ids(case_extra_data) assert result == [] def test_extract_ids_no_valid_issues_data(): """ GIVEN case_extra_data with issues.data that is not a list WHEN extract_ids is called THEN it returns empty list """ case_extra_data = { "case": { "aggregated_score": None, "assigned_user_mail": None, "assigned_user_pretty_name": None, "case_domain": "DOMAIN_POSTURE", "case_id": "62", }, "issues": {"data": {"id": "1"}}, } result = extract_ids(case_extra_data) assert result == [] def test_extract_ids_single_valid_item(): """ GIVEN case_extra_data with single valid issue WHEN extract_ids is called THEN it returns list with single issue_id """ case_extra_data = {"issues": {"data": [{"issue_id": "999", "description": "Single issue"}]}} result = extract_ids(case_extra_data) assert result == ["999"] def test_extract_ids(): """ Given: - json containing id When: - extract id from json Then: - check if id is extracted correctly """ case_extra_data_issue = { "case": { "aggregated_score": None, "assigned_user_mail": None, "assigned_user_pretty_name": None, "case_domain": "DOMAIN_POSTURE", "case_id": "62", "case_name": "caseName", "case_sources": ["CSPM Scanner"], "creation_time": 1761429680000, "critical_severity_issue_count": 0, "custom_fields": {}, "description": "caseDescription", "detection_time": None, "high_severity_issue_count": 2, "host_count": 0, "hosts": None, "is_blocked": False, "issue_categories": ["CONFIGURATION"], "issue_count": 2, "issues_grouping_status": "Enabled", "low_severity_issue_count": 0, "manual_description": None, "manual_score": None, "manual_severity": None, "med_severity_issue_count": 0, "mitre_tactics_ids_and_names": None, "mitre_techniques_ids_and_names": None, "modification_time": 1761429680000, "notes": None, "original_tags": ["tag1", "tag2"], "predicted_score": None, "resolve_comment": None, "resolved_timestamp": None, "rule_based_score": None, "severity": "high", "starred": False, "starred_manually": False, "status": "new", "tags": [], "user_count": 0, "users": [], "wildfire_hits": 0, "xdr_url": "https://example.com/incident-view?caseId=62", }, "issues": { "data": [ { "action": "SCANNED", "action_country": "UNKNOWN", "action_pretty": "Detected (Scanned)", "action_process_signature_status": "N/A", "actor_process_signature_status": "N/A", "agent_install_type": "NA", "agent_os_type": "NO_HOST", "attempt_counter": 0, "case_id": 62, "category": "CONFIGURATION", "causality_actor_process_signature_status": "N/A", "contains_featured_host": "NO", "contains_featured_ip": "NO", "contains_featured_user": "NO", "description": "description", "detection_timestamp": 1761464615083, "events_length": 1, "external_id": "externalId", "fw_is_phishing": "N/A", "is_pcap": False, "is_whitelisted": False, "issue_domain": "issueDomain", "issue_id": "282", "issue_type": "Unclassified", "last_modified_ts": 1761464616409, "local_insert_ts": 1761249196887, "matching_service_rule_id": "serviceRuleId", "matching_status": "UNMATCHABLE", "name": "Azure Storage Account default network access is set to 'Allow'", "original_tags": "DS:PANW/CSPM Scanner,DOM:Posture", "os_actor_process_signature_status": "N/A", "resolution_comment": "", "resolution_status": "STATUS_010_NEW", "severity": "high", "source": "CSPM Scanner", "starred": False, "tags": "DS:PANW/CSPM Scanner,DOM:Posture", }, { "action": "SCANNED", "action_country": "UNKNOWN", "action_pretty": "Detected (Scanned)", "action_process_signature_status": "N/A", "actor_process_signature_status": "N/A", "agent_install_type": "NA", "agent_os_type": "NO_HOST", "attempt_counter": 0, "case_id": 62, "category": "CONFIGURATION", "causality_actor_process_signature_status": "N/A", "contains_featured_host": "NO", "contains_featured_ip": "NO", "contains_featured_user": "NO", "description": "Azure Storage account Encryption Customer Managed Keys Disabled", "detection_timestamp": 1761464615083, "events_length": 1, "external_id": "P-987654321", "fw_is_phishing": "N/A", "is_pcap": False, "is_whitelisted": False, "issue_domain": "DOMAIN_POSTURE", "issue_id": "283", "issue_type": "Unclassified", "last_modified_ts": 1761464615986, "local_insert_ts": 1761249197825, "matching_status": "UNMATCHABLE", "name": "Azure Storage account Encryption Customer Managed Keys Disabled", "os_actor_process_signature_status": "N/A", "resolution_comment": "", "resolution_status": "STATUS_010_NEW", "severity": "high", "source": "CSPM Scanner", "starred": False, }, ], "total_count": 2, }, } id = extract_ids(case_extra_data_issue) assert id == ["282", "283"]