from copy import deepcopy import json import demistomock as demisto from GIBIncidentUpdateAllTypes import ( MAX_INCIDENTS, PAGE_SIZE, build_update_fields, get_gibid, iter_existing_incidents, main, ) # EntryType.ERROR is 4 in CommonServerPython; use the literal here so tests # stay independent of XSOAR runtime imports. ENTRY_TYPE_ERROR = 4 GIBID = "aaaaaaaabbbbccccddddeeeeeeeeeeeeeeeeee01" INCOMING_INCIDENT = { "id": "302", "name": "Test GIB Incident", "type": "GIB Compromised Card Group", "status": 1, "labels": [ {"type": "id", "value": GIBID}, {"type": "Brand", "value": "Group-IB Threat Intelligence & Attribution"}, ], "CustomFields": { "gibid": GIBID, "gibcredibility": 80, "gibseverity": "High", "gibportallink": f"https://example.com/cd/cards?id={GIBID}", }, } GET_INCIDENTS_FOUND = [{"Type": 1, "Contents": {"total": 1, "data": [{"id": "100", "gibid": GIBID}]}}] GET_INCIDENTS_EMPTY = [{"Type": 1, "Contents": {"total": 0, "data": []}}] GET_INCIDENTS_MULTIPLE = [ { "Type": 1, "Contents": { "total": 3, "data": [ {"id": "100", "gibid": GIBID}, {"id": "101", "gibid": GIBID}, {"id": "102", "gibid": GIBID}, ], }, } ] # --------------------------------------------------------------------------- # get_gibid # --------------------------------------------------------------------------- def test_gibid_from_custom_fields(): assert get_gibid({"CustomFields": {"gibid": GIBID}}) == GIBID def test_gibid_from_top_level(): assert get_gibid({"gibid": GIBID}) == GIBID def test_gibid_from_label_id(): assert get_gibid({"labels": [{"type": "id", "value": GIBID}]}) == GIBID def test_gibid_from_label_gibid(): assert get_gibid({"labels": [{"type": "gibid", "value": GIBID}]}) == GIBID def test_gibid_from_raw_json(): assert get_gibid({"rawJSON": json.dumps({"id": GIBID})}) == GIBID def test_gibid_missing(): assert get_gibid({"name": "test"}) is None def test_gibid_whitespace_only_is_missing(): assert get_gibid({"CustomFields": {"gibid": " "}}) is None # --------------------------------------------------------------------------- # iter_existing_incidents - streaming behavior # --------------------------------------------------------------------------- def _make_full_page(start: int = 0) -> list[dict]: return [{"id": str(i), "gibid": GIBID} for i in range(start, start + PAGE_SIZE)] def _wrap_page(data: list[dict]) -> list[dict]: return [{"Type": 1, "Contents": {"data": data}}] def test_iter_existing_query_shape(mocker): """First page request must carry the canonical query, sort and pagination.""" execute_command_mock = mocker.patch.object( demisto, "executeCommand", return_value=_wrap_page([{"id": "100", "gibid": GIBID}]) ) list(iter_existing_incidents(GIBID)) args = execute_command_mock.call_args.args assert args[0] == "getIncidents" assert args[1]["query"] == f"gibid: {GIBID} and -status:Closed" assert args[1]["sort"] == "created.desc" assert args[1]["size"] == PAGE_SIZE assert args[1]["page"] == 0 def test_iter_existing_stops_on_partial_page(mocker): """A page shorter than PAGE_SIZE means there are no further pages.""" execute_command_mock = mocker.patch.object( demisto, "executeCommand", return_value=_wrap_page([{"id": "100", "gibid": GIBID}]), ) result = list(iter_existing_incidents(GIBID)) assert result == [{"id": "100", "gibid": GIBID}] assert execute_command_mock.call_count == 1 def test_iter_existing_fetches_subsequent_pages_when_full(mocker): """A full PAGE_SIZE page must trigger a follow-up page request.""" pages = [ _wrap_page(_make_full_page(start=0)), _wrap_page([{"id": "999", "gibid": GIBID}]), ] execute_command_mock = mocker.patch.object(demisto, "executeCommand", side_effect=pages) result = list(iter_existing_incidents(GIBID)) assert execute_command_mock.call_count == 2 assert len(result) == PAGE_SIZE + 1 assert execute_command_mock.call_args_list[0].args[1]["page"] == 0 assert execute_command_mock.call_args_list[1].args[1]["page"] == 1 def test_iter_existing_respects_max_total_cap(mocker): """Iterator stops after `max_total` even if more data is available.""" pages = [_wrap_page(_make_full_page(start=0))] * 10 execute_command_mock = mocker.patch.object(demisto, "executeCommand", side_effect=pages) result = list(iter_existing_incidents(GIBID, max_total=50)) assert len(result) == 50 # 50 < PAGE_SIZE so a single API call is sufficient. assert execute_command_mock.call_count == 1 def test_iter_existing_max_total_across_multiple_pages(mocker): """`max_total` may straddle page boundaries; only the needed pages fetched.""" pages = [ _wrap_page(_make_full_page(start=0)), _wrap_page(_make_full_page(start=PAGE_SIZE)), _wrap_page(_make_full_page(start=PAGE_SIZE * 2)), ] execute_command_mock = mocker.patch.object(demisto, "executeCommand", side_effect=pages) result = list(iter_existing_incidents(GIBID, max_total=PAGE_SIZE + 5)) assert len(result) == PAGE_SIZE + 5 # Two pages are required: full first page + 5 from the second page. assert execute_command_mock.call_count == 2 def test_iter_existing_streams_lazily(mocker): """Generator must not eagerly load every page on construction.""" pages = [ _wrap_page(_make_full_page(start=0)), _wrap_page(_make_full_page(start=PAGE_SIZE)), ] execute_command_mock = mocker.patch.object(demisto, "executeCommand", side_effect=pages) iterator = iter_existing_incidents(GIBID) # Constructing the generator must NOT trigger any API call. assert execute_command_mock.call_count == 0 # Pulling the first item triggers exactly one API call. next(iterator) assert execute_command_mock.call_count == 1 # Draining page 0 still costs one API call. for _ in range(PAGE_SIZE - 1): next(iterator) assert execute_command_mock.call_count == 1 # Item PAGE_SIZE+1 forces the second page fetch. next(iterator) assert execute_command_mock.call_count == 2 def test_iter_existing_handles_error_response_gracefully(mocker): """An XSOAR error response must terminate iteration without raising.""" error_response = [{"Type": ENTRY_TYPE_ERROR, "Contents": "boom"}] mocker.patch.object(demisto, "executeCommand", return_value=error_response) assert list(iter_existing_incidents(GIBID)) == [] def test_iter_existing_handles_empty_response(mocker): mocker.patch.object(demisto, "executeCommand", return_value=GET_INCIDENTS_EMPTY) assert list(iter_existing_incidents(GIBID)) == [] def test_iter_existing_handles_none_response(mocker): mocker.patch.object(demisto, "executeCommand", return_value=None) assert list(iter_existing_incidents(GIBID)) == [] def test_iter_existing_invalid_max_total_returns_nothing(mocker): execute_command_mock = mocker.patch.object(demisto, "executeCommand") assert list(iter_existing_incidents(GIBID, max_total=0)) == [] assert list(iter_existing_incidents(GIBID, page_size=0)) == [] assert execute_command_mock.call_count == 0 # --------------------------------------------------------------------------- # build_update_fields # --------------------------------------------------------------------------- def test_build_update_fields_flattens_custom_fields_and_propagates_all_other_fields(): """All incoming fields propagate; only `id` and the `CustomFields` container itself are stripped. `id` is excluded because it is the target identifier for `setIncident` (passing it would silently redirect the update to the incoming incident). `CustomFields` is excluded as a *container* because its members are flattened into top-level kwargs. """ incident = deepcopy(INCOMING_INCIDENT) incident["occurred"] = "2025-01-01T00:00:00Z" incident["rawJSON"] = '{"id":"abc"}' update_fields = build_update_fields(incident) # Hard guards: protect `setIncident` call correctness. assert "id" not in update_fields assert "CustomFields" not in update_fields # All previously-skipped XSOAR fields now propagate as-is. assert update_fields["labels"] == incident["labels"] assert update_fields["occurred"] == "2025-01-01T00:00:00Z" assert update_fields["rawJSON"] == '{"id":"abc"}' assert update_fields["type"] == "GIB Compromised Card Group" assert update_fields["status"] == 1 # CustomFields are flattened into top-level keys. assert update_fields["gibid"] == GIBID assert update_fields["gibcredibility"] == 80 assert update_fields["gibseverity"] == "High" assert update_fields["name"] == "Test GIB Incident" def test_build_update_fields_strips_id_from_custom_fields_too(): """If `id` ever leaks into CustomFields, it must still be filtered out.""" incident = { "id": "incoming-1", "name": "x", "CustomFields": {"id": "leaked", "gibid": GIBID}, } update_fields = build_update_fields(incident) assert "id" not in update_fields def test_build_update_fields_drops_none_values(): incident = {"name": "x", "details": None, "CustomFields": {"gibid": GIBID, "gibsource": None}} update_fields = build_update_fields(incident) assert "details" not in update_fields assert "gibsource" not in update_fields assert update_fields == {"name": "x", "gibid": GIBID} # --------------------------------------------------------------------------- # main # --------------------------------------------------------------------------- def test_main_no_gibid_returns_true(mocker): mocker.patch.object(demisto, "incident", return_value={"name": "test"}) mock_results = mocker.patch("GIBIncidentUpdateAllTypes.return_results") main() mock_results.assert_called_once_with(True) def test_main_no_duplicate_returns_true(mocker): mocker.patch.object(demisto, "incident", return_value=deepcopy(INCOMING_INCIDENT)) mocker.patch.object(demisto, "executeCommand", return_value=GET_INCIDENTS_EMPTY) mock_results = mocker.patch("GIBIncidentUpdateAllTypes.return_results") main() mock_results.assert_called_once_with(True) def test_main_duplicate_found_calls_setincident_and_returns_false(mocker): mocker.patch.object(demisto, "incident", return_value=deepcopy(INCOMING_INCIDENT)) def mock_execute_command(command, args): if command == "getIncidents": return GET_INCIDENTS_FOUND return [{"Type": 1, "Contents": "ok"}] execute_command_mock = mocker.patch.object(demisto, "executeCommand", side_effect=mock_execute_command) mock_results = mocker.patch("GIBIncidentUpdateAllTypes.return_results") main() mock_results.assert_called_once_with(False) set_calls = [c for c in execute_command_mock.call_args_list if c.args[0] == "setIncident"] assert len(set_calls) == 1 payload = set_calls[0].args[1] assert payload["id"] == "100" assert payload["gibid"] == GIBID assert payload["name"] == "Test GIB Incident" assert payload["gibseverity"] == "High" def test_main_updates_all_open_duplicates(mocker): mocker.patch.object(demisto, "incident", return_value=deepcopy(INCOMING_INCIDENT)) def mock_execute_command(command, args): if command == "getIncidents": return GET_INCIDENTS_MULTIPLE return [{"Type": 1, "Contents": "ok"}] execute_command_mock = mocker.patch.object(demisto, "executeCommand", side_effect=mock_execute_command) mocker.patch("GIBIncidentUpdateAllTypes.return_results") main() set_calls = [c for c in execute_command_mock.call_args_list if c.args[0] == "setIncident"] assert [c.args[1]["id"] for c in set_calls] == ["100", "101", "102"] def test_main_uses_existing_id_not_incoming(mocker): """The setIncident call must target the duplicate's id, not the incoming one.""" mocker.patch.object(demisto, "incident", return_value=deepcopy(INCOMING_INCIDENT)) def mock_execute_command(command, args): if command == "getIncidents": return GET_INCIDENTS_FOUND return [{"Type": 1, "Contents": "ok"}] execute_command_mock = mocker.patch.object(demisto, "executeCommand", side_effect=mock_execute_command) mocker.patch("GIBIncidentUpdateAllTypes.return_results") main() set_call = next(c for c in execute_command_mock.call_args_list if c.args[0] == "setIncident") assert set_call.args[1]["id"] == "100" def test_main_does_not_mutate_incident(mocker): incident = deepcopy(INCOMING_INCIDENT) original = deepcopy(incident) mocker.patch.object(demisto, "incident", return_value=incident) def mock_execute_command(command, args): if command == "getIncidents": return GET_INCIDENTS_FOUND return [{"Type": 1, "Contents": "ok"}] mocker.patch.object(demisto, "executeCommand", side_effect=mock_execute_command) mocker.patch("GIBIncidentUpdateAllTypes.return_results") main() assert incident == original def test_main_streams_pages_without_loading_all_into_memory(mocker): """Drives `main` against PAGE_SIZE+1 duplicates and verifies that the fetch and the update phases are interleaved (page-streamed), i.e. the very first setIncident call happens before the second getIncidents page is fetched. """ mocker.patch.object(demisto, "incident", return_value=deepcopy(INCOMING_INCIDENT)) page_zero = _wrap_page(_make_full_page(start=0)) page_one = _wrap_page([{"id": str(PAGE_SIZE), "gibid": GIBID}]) set_incident_ok = [{"Type": 1, "Contents": "ok"}] call_log: list[str] = [] def mock_execute_command(command, args): if command == "getIncidents": page_index = args["page"] call_log.append(f"getIncidents:{page_index}") return page_zero if page_index == 0 else page_one if command == "setIncident": call_log.append(f"setIncident:{args['id']}") return set_incident_ok raise AssertionError(f"unexpected command: {command}") mocker.patch.object(demisto, "executeCommand", side_effect=mock_execute_command) mocker.patch("GIBIncidentUpdateAllTypes.return_results") main() set_events = [e for e in call_log if e.startswith("setIncident:")] assert len(set_events) == PAGE_SIZE + 1 # Streaming guarantee: the first setIncident call must happen BEFORE # the second getIncidents page is fetched. This is the property that # keeps memory usage at O(PAGE_SIZE) instead of O(MAX_INCIDENTS). first_set_idx = next(i for i, e in enumerate(call_log) if e.startswith("setIncident:")) second_get_idx = next(i for i, e in enumerate(call_log) if e == "getIncidents:1") assert first_set_idx < second_get_idx, ( "setIncident calls must be interleaved with getIncidents pages " "so that memory usage stays bounded by PAGE_SIZE; observed call " f"log: {call_log}" ) def test_main_max_incidents_circuit_breaker_is_strict(mocker): """`main` must never call setIncident more than MAX_INCIDENTS times.""" mocker.patch.object(demisto, "incident", return_value=deepcopy(INCOMING_INCIDENT)) full_page = _wrap_page(_make_full_page(start=0)) def mock_execute_command(command, args): if command == "getIncidents": return full_page # endless supply return [{"Type": 1, "Contents": "ok"}] execute_command_mock = mocker.patch.object(demisto, "executeCommand", side_effect=mock_execute_command) mocker.patch("GIBIncidentUpdateAllTypes.return_results") main() set_calls = [c for c in execute_command_mock.call_args_list if c.args[0] == "setIncident"] assert len(set_calls) <= MAX_INCIDENTS def test_main_setincident_failure_does_not_abort_remaining_updates(mocker): mocker.patch.object(demisto, "incident", return_value=deepcopy(INCOMING_INCIDENT)) def mock_execute_command(command, args): if command == "getIncidents": return GET_INCIDENTS_MULTIPLE if args["id"] == "101": return [{"Type": ENTRY_TYPE_ERROR, "Contents": "boom"}] return [{"Type": 1, "Contents": "ok"}] execute_command_mock = mocker.patch.object(demisto, "executeCommand", side_effect=mock_execute_command) mock_results = mocker.patch("GIBIncidentUpdateAllTypes.return_results") main() set_calls = [c for c in execute_command_mock.call_args_list if c.args[0] == "setIncident"] # All three duplicates were attempted, even though the middle one failed. assert [c.args[1]["id"] for c in set_calls] == ["100", "101", "102"] # The incoming incident is still dropped because real duplicates exist. mock_results.assert_called_once_with(False)