"""Unit tests for the Proofpoint Cloud Threat Response integration.""" import json from pathlib import Path from typing import Any import pytest from ProofpointCloudThreatResponse import ( Client, ProofpointCTRAuthHandler, build_filters_body, fetch_incidents, format_ctr_date, parse_ctr_date, proofpoint_ctr_incident_get_command, proofpoint_ctr_incidents_list_command, run_test_module, ) TEST_DATA = Path(__file__).parent / "test_data" BASE_URL = "https://threatprotection-api.proofpoint.com" def _load(name: str) -> dict: with (TEST_DATA / name).open() as fp: return json.load(fp) @pytest.fixture() def _patch_context(mocker): """Patch context-store I/O so the auth handler never hits the runtime.""" mocker.patch.object(ProofpointCTRAuthHandler, "_load_token_from_context") mocker.patch.object(ProofpointCTRAuthHandler, "_save_token_to_context") @pytest.fixture() def client(_patch_context, mocker) -> Client: """Return a Client whose ``_http_request`` is mock-able per test.""" # Avoid initializing the underlying ContentClient (httpx etc.) - we mock the # I/O surface exposed by Client.list_incidents / Client.get_incident. mocker.patch("ProofpointCloudThreatResponse.ContentClient.__init__", return_value=None) instance = Client.__new__(Client) # Provide just enough state for the test: instance.timeout = 60.0 # type: ignore[attr-defined] return instance # --------------------------------------------------------------------------- helpers def test_format_ctr_date_strips_timezone(): from datetime import datetime formatted = format_ctr_date(datetime(2024, 11, 26, 16, 18, 7)) assert formatted == "2024-11-26 16:18:07" def test_parse_ctr_date_handles_freetext(): parsed = parse_ctr_date("2024-11-26 16:18:07") assert parsed is not None assert parsed.year == 2024 assert parsed.month == 11 def test_build_filters_body_omits_empty_filters(): from datetime import datetime body = build_filters_body( start_time=datetime(2024, 11, 26, 16, 18, 7), end_time=datetime(2024, 11, 26, 16, 19, 7), end_row=10, ) assert body == { "filters": { "time_range_filter": { "start": "2024-11-26 16:18:07", "end": "2024-11-26 16:19:07", }, }, "startRow": 0, "endRow": 10, "sortParams": [{"sort": "desc", "colId": "createdAt"}], } def test_build_filters_body_validates_allowed_values(): with pytest.raises(Exception, match="source_filters"): build_filters_body(source_filters=["bogus"]) # --------------------------------------------------------------------------- auth def test_auth_handler_rejects_empty_credentials(): from ProofpointCloudThreatResponse import ProofpointCTRAuthHandler with pytest.raises(Exception, match="Client ID"): ProofpointCTRAuthHandler(client_id="", client_secret="x") with pytest.raises(Exception, match="Client Secret"): ProofpointCTRAuthHandler(client_id="x", client_secret="") def test_auth_handler_token_validity(mocker): import time as _time mocker.patch.object(ProofpointCTRAuthHandler, "_load_token_from_context") handler = ProofpointCTRAuthHandler(client_id="id", client_secret="secret") handler._access_token = "abc" handler._expires_at = int(_time.time()) + 3600 assert handler._token_is_valid() is True handler._expires_at = int(_time.time()) - 1 assert handler._token_is_valid() is False handler._access_token = None handler._expires_at = int(_time.time()) + 3600 assert handler._token_is_valid() is False # --------------------------------------------------------------------------- commands def test_list_incidents_command_builds_output(client: Client, mocker): mocker.patch.object(Client, "list_incidents", return_value=_load("incidents_list.json")) result = proofpoint_ctr_incidents_list_command( client, {"limit": "10", "source_filters": "abuse_mailbox"}, ) assert result.outputs_prefix == "ProofPointCloud.Incident" assert len(result.outputs) == 2 assert result.outputs[0]["id"] == "00000000-0000-0000-0000-000000000001" assert "Suspicious phishing attempt" in result.readable_output def test_list_incidents_rejects_invalid_filter(client: Client): with pytest.raises(Exception, match="other_filters"): proofpoint_ctr_incidents_list_command(client, {"other_filters": "bogus"}) def test_get_incident_command_iterates_ids(client: Client, mocker): calls: list[str] = [] def _fake_get(self, incident_id: str) -> dict[str, Any]: calls.append(incident_id) return _load("incident_get.json") mocker.patch.object(Client, "get_incident", _fake_get) result = proofpoint_ctr_incident_get_command(client, {"incident_id": "aaa,bbb"}) assert calls == ["aaa", "bbb"] assert len(result.outputs) == 2 # The get command flattens the summary fields to the top level so that # outputs_key_field="id" matches the list command and XSOAR can enrich # the existing context entry rather than creating a duplicate. assert result.outputs[0]["displayId"] == 12345 def test_get_incident_requires_id(client: Client): with pytest.raises(Exception, match="incident_id"): proofpoint_ctr_incident_get_command(client, {}) # --------------------------------------------------------------------------- test_module def test_test_module_ok(client: Client, mocker): mocker.patch.object(Client, "list_incidents", return_value={"incidents": []}) assert run_test_module(client, {"isFetch": False}) == "ok" def test_test_module_rejects_both_states(client: Client): msg = run_test_module( client, {"isFetch": True, "fetch_states": "open_incidents,closed_incidents"}, ) assert "empty result" in msg def test_test_module_requires_state_when_fetching(client: Client): msg = run_test_module(client, {"isFetch": True, "fetch_states": ""}) assert "must select at least one" in msg # --------------------------------------------------------------------------- fetch def test_fetch_incidents_first_run(client: Client, mocker): mocker.patch.object(Client, "list_incidents", return_value=_load("incidents_list.json")) mocker.patch.object(Client, "get_incident", return_value=_load("incident_get.json")) next_run, incidents = fetch_incidents( client, { "first_fetch": "3 days", "max_fetch": "50", "fetch_delta": "1", "fetch_states": "open_incidents", }, last_run={}, ) assert len(incidents) == 2 assert incidents[0]["dbotMirrorId"] == "00000000-0000-0000-0000-000000000001" assert "last_fetch" in next_run assert "last_fetched_ids" in next_run # last_fetch must be the createdAt of the latest incident (incident 1: 2024-01-01T10:00:00) assert next_run["last_fetch"] == "2024-01-01 10:00:00" def test_fetch_incidents_dedupes_seen_ids(client: Client, mocker): mocker.patch.object(Client, "list_incidents", return_value=_load("incidents_list.json")) mocker.patch.object(Client, "get_incident", return_value=_load("incident_get.json")) next_run, incidents = fetch_incidents( client, { "first_fetch": "3 days", "max_fetch": "50", "fetch_delta": "1", "fetch_states": "open_incidents", }, last_run={ "last_fetch": "2024-01-01 08:00:00", "last_fetched_ids": ["00000000-0000-0000-0000-000000000001"], }, ) assert len(incidents) == 1 assert incidents[0]["dbotMirrorId"] == "00000000-0000-0000-0000-000000000002" # last_fetch should advance to the createdAt of the latest new incident (incident 2) assert next_run["last_fetch"] == "2024-01-01 09:00:00" def test_fetch_incidents_rejects_both_states(client: Client): with pytest.raises(Exception, match="empty result"): fetch_incidents( client, { "first_fetch": "3 days", "fetch_states": "open_incidents,closed_incidents", }, last_run={}, ) def test_fetch_incidents_caps_max_fetch(client: Client, mocker): captured: dict = {} def _capture(self, body): captured.update(body) return {"incidents": []} mocker.patch.object(Client, "list_incidents", _capture) fetch_incidents( client, { "first_fetch": "3 days", "max_fetch": "9999", "fetch_delta": "0", "fetch_states": "open_incidents", }, last_run={}, ) # max_fetch is capped at MAX_PAGE_SIZE (200) -> endRow = 200 (exclusive upper bound) assert captured["endRow"] == 200 assert captured["startRow"] == 0 assert captured["sortParams"] == [{"sort": "asc", "colId": "createdAt"}] assert captured["filters"]["other_filters"] == ["open_incidents"]