revert API use

Author: mariot

palo-alto-cortex-xsoar/src/services/client_api.py

@@ -1,70 +1,55 @@
-from http.cookiejar import DefaultCookiePolicy
+import random
+import uuid
+from datetime import datetime, timezone
 from typing import Optional
 
-from requests import Session
-from requests.adapters import HTTPAdapter
-from requests.packages.urllib3.util import Retry
 from src.models.authentication import Authentication
 from src.models.incident import XSOARSearchIncidentsResponse
 
-REQUESTS_TIMEOUT_SECONDS = 60
-
 
 class PaloAltoCortexXSOARClientAPI:
     def __init__(self, auth: Authentication, api_url: str) -> None:
         self._auth = auth
         self.api_url = api_url
 
-        self.session = self._prepare_session()
-
-    def _build_url(self, path: str) -> str:
-        """Build a full URL from the configured api_url and a path."""
-        return f"{self.api_url.rstrip('/')}{path}"
-
-    def _prepare_session(self) -> Session:
-        """Preparing a session with automatic retries (with increasing backoff) and no cookies"""
-        retries = Retry(
-            total=5,
-            allowed_methods=["POST"],
-            status_forcelist=[429, 500, 502, 503, 504],
-            backoff_factor=0.5,
-            backoff_jitter=0.2,
-        )
-        s = Session()
-        s.mount("http://", HTTPAdapter(max_retries=retries))
-        s.mount("https://", HTTPAdapter(max_retries=retries))
-        s.cookies.set_policy(DefaultCookiePolicy(allowed_domains=[]))
-        return s
-
     def search_incidents(
         self,
         from_date: Optional[str] = None,
         to_date: Optional[str] = None,
         search_from: int = 0,
         search_to: int = 100,
     ) -> XSOARSearchIncidentsResponse:
-        url = self._build_url("/xsoar/public/v1/incidents/search")
-        headers = self._auth.get_headers()
-
-        size = search_to - search_from
-        page = search_from // size if size > 0 else 0
-
-        body = {
-            "filter": {
-                "page": page,
-                "size": size,
-                "sort": [{"field": "created", "asc": True}],
-            }
-        }
-
-        if from_date:
-            body["filter"]["fromDate"] = from_date
-
-        if to_date:
-            body["filter"]["toDate"] = to_date
-
-        response = self.session.post(
-            url, headers=headers, json=body, timeout=REQUESTS_TIMEOUT_SECONDS
+        _ = (from_date, to_date, search_from, search_to)
+
+        incident_count = random.randint(1, 3)
+        detection_timestamp = int(datetime.now(timezone.utc).timestamp() * 1000)
+
+        data = []
+        for _ in range(incident_count):
+            data.append(
+                {
+                    "id": str(uuid.uuid4()),
+                    "name": "Dummy XSOAR Incident",
+                    "CustomFields": {
+                        "xdralerts": [
+                            {
+                                "alert_id": str(uuid.uuid4()),
+                                "case_id": random.randint(1, 1000),
+                                "action_pretty": random.choice(
+                                    ["Detected (Reported)", "Prevented (Blocked)"]
+                                ),
+                                "actor_process_command_line": (
+                                    f"oaev-implant-{uuid.uuid4()}-agent-{uuid.uuid4()}"
+                                ),
+                                "actor_process_image_name": "oaev-implant.exe",
+                                "actor_process_image_path": "C:/Dummy/oaev-implant.exe",
+                                "detection_timestamp": detection_timestamp,
+                            }
+                        ]
+                    },
+                }
+            )
+
+        return XSOARSearchIncidentsResponse.model_validate(
+            {"total": len(data), "data": data}
         )
-        response.raise_for_status()
-        return XSOARSearchIncidentsResponse.model_validate(response.json())

palo-alto-cortex-xsoar/tests/test_client_api.py

@@ -1,7 +1,4 @@
-from unittest.mock import patch
-
 import pytest
-import requests
 from src.models.authentication import Authentication
 from src.models.incident import XSOARSearchIncidentsResponse
 from src.services.client_api import PaloAltoCortexXSOARClientAPI
@@ -17,187 +14,49 @@ def api_client(auth):
     return PaloAltoCortexXSOARClientAPI(auth=auth, api_url="https://test.xsoar.com")
 
 
-def test_search_incidents_success(api_client):
-    mock_response = {
-        "total": 1,
-        "data": [
-            {
-                "id": "1",
-                "name": "Test Incident",
-                "CustomFields": {
-                    "xdralerts": [
-                        {"alert_id": "alert1", "detection_timestamp": 1600000000000}
-                    ]
-                },
-            }
-        ],
-    }
-
-    with patch("requests.Session.post") as mock_post:
-        mock_post.return_value.json.return_value = mock_response
-        mock_post.return_value.status_code = 200
-
-        response = api_client.search_incidents(
-            from_date="2023-01-01T00:00:00Z",
-            to_date="2023-01-01T23:59:59Z",
-            search_from=0,
-            search_to=10,
-        )
-
-        assert isinstance(response, XSOARSearchIncidentsResponse)
-        assert response.total == 1
-        assert len(response.data) == 1
-        assert response.data[0].id == "1"
-        assert response.data[0].custom_fields.xdralerts[0].alert_id == "alert1"
-
-        mock_post.assert_called_once()
-        args, kwargs = mock_post.call_args
-        assert args[0] == "https://test.xsoar.com/xsoar/public/v1/incidents/search"
-        assert kwargs["json"]["filter"]["size"] == 10
-        assert kwargs["json"]["filter"]["page"] == 0
-        assert kwargs["json"]["filter"]["fromDate"] == "2023-01-01T00:00:00Z"
-        assert kwargs["json"]["filter"]["toDate"] == "2023-01-01T23:59:59Z"
-
-
-def test_search_incidents_http_error(api_client):
-    with patch("requests.Session.post") as mock_post:
-        mock_post.return_value.raise_for_status.side_effect = (
-            requests.exceptions.HTTPError("Error")
-        )
-
-        with pytest.raises(requests.exceptions.HTTPError):
-            api_client.search_incidents()
-
-
-def test_search_incidents_pagination(api_client):
-    with patch("requests.Session.post") as mock_post:
-        mock_post.return_value.json.return_value = {"total": 0, "data": []}
-
-        api_client.search_incidents(search_from=20, search_to=30)
-
-        _, kwargs = mock_post.call_args
-        assert kwargs["json"]["filter"]["size"] == 10
-        assert kwargs["json"]["filter"]["page"] == 2
-
-
-# --- New tests ---
-
-
-def test_search_incidents_no_dates(api_client):
-    """When no dates are provided, fromDate and toDate are absent."""
-    with patch("requests.Session.post") as mock_post:
-        mock_post.return_value.json.return_value = {"total": 0, "data": []}
-        api_client.search_incidents()
-        _, kwargs = mock_post.call_args
-        assert "fromDate" not in kwargs["json"]["filter"]
-        assert "toDate" not in kwargs["json"]["filter"]
-
-
-def test_search_incidents_only_from_date(api_client):
-    """When only from_date is provided, toDate is absent."""
-    with patch("requests.Session.post") as mock_post:
-        mock_post.return_value.json.return_value = {"total": 0, "data": []}
-        api_client.search_incidents(from_date="2026-01-01T00:00:00Z")
-        _, kwargs = mock_post.call_args
-        assert kwargs["json"]["filter"]["fromDate"] == "2026-01-01T00:00:00Z"
-        assert "toDate" not in kwargs["json"]["filter"]
-
-
-def test_search_incidents_only_to_date(api_client):
-    """When only to_date is provided, fromDate is absent."""
-    with patch("requests.Session.post") as mock_post:
-        mock_post.return_value.json.return_value = {"total": 0, "data": []}
-        api_client.search_incidents(to_date="2026-12-31T23:59:59Z")
-        _, kwargs = mock_post.call_args
-        assert "fromDate" not in kwargs["json"]["filter"]
-        assert kwargs["json"]["filter"]["toDate"] == "2026-12-31T23:59:59Z"
-
-
-def test_search_incidents_zero_size(api_client):
-    """When search_from == search_to, size is 0 and page is 0."""
-    with patch("requests.Session.post") as mock_post:
-        mock_post.return_value.json.return_value = {"total": 0, "data": []}
-        api_client.search_incidents(search_from=5, search_to=5)
-        _, kwargs = mock_post.call_args
-        assert kwargs["json"]["filter"]["size"] == 0
-        assert kwargs["json"]["filter"]["page"] == 0
-
-
-def test_search_incidents_default_page_size(api_client):
-    """Default search_from=0, search_to=100 gives size=100, page=0."""
-    with patch("requests.Session.post") as mock_post:
-        mock_post.return_value.json.return_value = {"total": 0, "data": []}
-        api_client.search_incidents()
-        _, kwargs = mock_post.call_args
-        assert kwargs["json"]["filter"]["size"] == 100
-        assert kwargs["json"]["filter"]["page"] == 0
-
-
-def test_search_incidents_headers_sent(api_client, auth):
-    """Auth headers are included in the request."""
-    expected_headers = auth.get_headers()
-    with patch("requests.Session.post") as mock_post:
-        mock_post.return_value.json.return_value = {"total": 0, "data": []}
-        api_client.search_incidents()
-        _, kwargs = mock_post.call_args
-        assert kwargs["headers"] == expected_headers
-
-
-def test_search_incidents_sort_order(api_client):
-    """Request body always includes sort by 'created' ascending."""
-    with patch("requests.Session.post") as mock_post:
-        mock_post.return_value.json.return_value = {"total": 0, "data": []}
-        api_client.search_incidents()
-        _, kwargs = mock_post.call_args
-        assert kwargs["json"]["filter"]["sort"] == [{"field": "created", "asc": True}]
-
-
-def test_search_incidents_connection_error(api_client):
-    """Connection errors propagate."""
-    with patch(
-        "requests.Session.post",
-        side_effect=requests.exceptions.ConnectionError("no route"),
-    ):
-        with pytest.raises(requests.exceptions.ConnectionError):
-            api_client.search_incidents()
-
-
-def test_search_incidents_timeout(api_client):
-    """Timeout errors propagate."""
-    with patch(
-        "requests.Session.post", side_effect=requests.exceptions.Timeout("timed out")
-    ):
-        with pytest.raises(requests.exceptions.Timeout):
-            api_client.search_incidents()
-
-
-def test_search_incidents_multiple_incidents(api_client):
-    """Response with multiple incidents is parsed correctly."""
-    mock_response = {
-        "total": 2,
-        "data": [
-            {
-                "id": "1",
-                "name": "Inc 1",
-                "CustomFields": {
-                    "xdralerts": [{"alert_id": "a1", "detection_timestamp": 1000}]
-                },
-            },
-            {
-                "id": "2",
-                "name": "Inc 2",
-                "CustomFields": {
-                    "xdralerts": [{"alert_id": "a2", "detection_timestamp": 2000}]
-                },
-            },
-        ],
-    }
-    with patch("requests.Session.post") as mock_post:
-        mock_post.return_value.json.return_value = mock_response
-        response = api_client.search_incidents()
-        assert response.total == 2
-        assert len(response.data) == 2
-        assert response.data[1].custom_fields.xdralerts[0].alert_id == "a2"
+def test_search_incidents_returns_valid_dummy_response(api_client):
+    response = api_client.search_incidents()
+
+    assert isinstance(response, XSOARSearchIncidentsResponse)
+    assert response.total == len(response.data)
+    assert response.total >= 1
+
+    for incident in response.data:
+        assert incident.id
+        assert incident.custom_fields is not None
+        assert len(incident.custom_fields.xdralerts) == 1
+        alert = incident.custom_fields.xdralerts[0]
+        assert alert.alert_id
+        assert isinstance(alert.detection_timestamp, int)
+
+
+def test_search_incidents_accepts_filters_without_external_calls(api_client):
+    response = api_client.search_incidents(
+        from_date="2026-01-01T00:00:00Z",
+        to_date="2026-01-01T23:59:59Z",
+        search_from=10,
+        search_to=20,
+    )
+
+    assert isinstance(response, XSOARSearchIncidentsResponse)
+    assert response.total == len(response.data)
+
+
+def test_search_incidents_can_be_forced_to_multiple_items(api_client, monkeypatch):
+    monkeypatch.setattr("src.services.client_api.random.randint", lambda a, b: 2)
+    response = api_client.search_incidents()
+    assert response.total == 2
+    assert len(response.data) == 2
+
+
+def test_search_incidents_generated_ids_are_unique(api_client):
+    response = api_client.search_incidents()
+    incident_ids = [incident.id for incident in response.data]
+    alert_ids = [
+        incident.custom_fields.xdralerts[0].alert_id for incident in response.data
+    ]
+    assert len(set(incident_ids)) == len(incident_ids)
+    assert len(set(alert_ids)) == len(alert_ids)
 
 
 def test_api_client_stores_api_url():