Ciac 11100 playbook run in build

.changelog/4658.yml

@@ -0,0 +1,4 @@
+changes:
+- description: Adding new command **test-use-case** to test use case flows on cloud machines.
+  type: internal
+pr_number: 4658

conftest.py

@@ -175,3 +175,12 @@ def disable_log_colors():
 @pytest.fixture(autouse=True)
 def clear_cache():
     tools.get_file.cache_clear()
+
+
+def pytest_addoption(parser):
+    parser.addoption(
+        "--client_conf",
+        action="store",
+        default=None,
+        help="Custom client configuration",
+    )

demisto_sdk/__main__.py

@@ -67,6 +67,9 @@
 from demisto_sdk.commands.test_content.test_modeling_rule.modeling_rules_setup import (
     modeling_rules_app,
 )
+from demisto_sdk.commands.test_content.test_use_case.test_use_case import (
+    run_test_use_case,
+)
 from demisto_sdk.commands.update_release_notes.update_release_notes_setup import (
     update_release_notes,
 )
@@ -203,6 +206,9 @@
 app.command(name="generate-modeling-rules", help="Generated modeling-rules.")(
     generate_modeling_rules
 )
+app.command(
+    name="test-use-case", hidden=True, no_args_is_help=True, help="Test Use Cases."
+)(run_test_use_case)
 app.command(
     name="lint", help="Deprecated, use demisto-sdk pre-commit instead.", hidden=True
 )(lint)

demisto_sdk/commands/common/clients/__init__.py

@@ -27,6 +27,7 @@
     DEMISTO_PASSWORD,
     DEMISTO_USERNAME,
     DEMISTO_VERIFY_SSL,
+    LCAS_ID,
     MarketplaceVersions,
 )
 from demisto_sdk.commands.common.logger import logger
@@ -132,6 +133,7 @@ def get_client_from_server_type(
     auth_id: Optional[str] = None,
     username: Optional[str] = None,
     password: Optional[str] = None,
+    lcas_id: Optional[str] = None,
     verify_ssl: Optional[bool] = None,
     raise_if_server_not_healthy: bool = True,
 ) -> XsoarClient:
@@ -144,6 +146,7 @@ def get_client_from_server_type(
         auth_id: the auth ID, if not provided will take from XSIAM_AUTH_ID env var
         username: the username to authenticate, relevant only for xsoar on prem
         password: the password to authenticate, relevant only for xsoar on prem
+        lcas_id: the lcas id of the current cloud machine.
         verify_ssl: whether in each request SSL should be verified, True if yes, False if not,
                     if verify_ssl = None, will take the SSL verification from DEMISTO_VERIFY_SSL env var
         raise_if_server_not_healthy: whether to raise an exception if the server is not healthy
@@ -156,6 +159,7 @@ def get_client_from_server_type(
     _auth_id = auth_id or os.getenv(AUTH_ID)
     _username = username or os.getenv(DEMISTO_USERNAME, "")
     _password = password or os.getenv(DEMISTO_PASSWORD, "")
+    _lcas_id = lcas_id or os.getenv(LCAS_ID, "")
     _verify_ssl = (
         verify_ssl
         if verify_ssl is not None
@@ -188,6 +192,7 @@ def get_client_from_server_type(
                 api_key=_api_key,
                 auth_id=_auth_id,
                 verify_ssl=_verify_ssl,
+                lcas_id=_lcas_id,
             ),
             should_validate_server_type=should_validate_server_type,
             raise_if_server_not_healthy=raise_if_server_not_healthy,
@@ -202,6 +207,7 @@ def get_client_from_server_type(
                 api_key=_api_key,
                 auth_id=_auth_id,
                 verify_ssl=_verify_ssl,
+                lcas_id=_lcas_id,
             ),
             should_validate_server_type=should_validate_server_type,
             raise_if_server_not_healthy=raise_if_server_not_healthy,
@@ -232,3 +238,31 @@ def get_client_from_server_type(
             f"make sure the {DEMISTO_BASE_URL}, {DEMISTO_KEY}, {AUTH_ID} are defined properly"
         )
         raise
+
+
+# =================== Playbook Flow Tests =================
+
+
+def parse_str_to_dict(input_str):
+    """Internal function to convert a string representing a dictionary into an actual dictionary.
+
+    Args:
+        input_str (str): A string in the format 'key1=value1,key2=value2'.
+
+    Returns:
+        dict: A dictionary with the parsed key-value pairs.
+    """
+    x = dict(pair.split("=") for pair in input_str.split(",") if "=" in pair)
+    logger.info(x.get("base_url", "no base url"))
+    return dict(pair.split("=") for pair in input_str.split(",") if "=" in pair)
+
+
+def get_client_conf_from_pytest_request(request):
+    # Manually parse command-line argument
+    for arg in request.config.invocation_params.args:
+        if isinstance(arg, str) and arg.startswith("--client_conf="):
+            logger.info("there is --client_conf recognized")
+            client_conf = arg.replace("--client_conf=", "")
+            return parse_str_to_dict(client_conf)
+    # If a client data was not provided, we proceed to use default.
+    return None

demisto_sdk/commands/common/clients/configs.py

@@ -10,6 +10,7 @@
     DEMISTO_PASSWORD,
     DEMISTO_USERNAME,
     DEMISTO_VERIFY_SSL,
+    LCAS_ID,
     XSIAM_COLLECTOR_TOKEN,
     XSIAM_TOKEN,
 )
@@ -72,6 +73,7 @@ def __eq__(self, other):
 
 class XsoarSaasClientConfig(XsoarClientConfig):
     auth_id: str = Field(default=os.getenv(AUTH_ID), description="XSOAR/XSIAM Auth ID")
+    lcas_id: str = Field(default=os.getenv(LCAS_ID), description="XSOAR/XSIAM LCAS ID")
 
     @root_validator()
     def validate_auth_params(cls, values: Dict[str, Any]):

demisto_sdk/commands/common/clients/xsiam/xsiam_api_client.py

@@ -46,6 +46,31 @@ def server_type(self) -> ServerType:
     def marketplace(self) -> MarketplaceVersions:
         return MarketplaceVersions.MarketplaceV2
 
+    """
+    #############################
+    Helper methods
+    #############################
+    """
+
+    def _process_response(self, response, status_code, expected_status=200):
+        """Process the HTTP response coming from the XSOAR client."""
+        if status_code == expected_status:
+            if response:
+                try:
+                    return json.loads(response)
+                except json.JSONDecodeError:
+                    api_response = (
+                        response.replace("'", '"')
+                        .replace("False", "false")
+                        .replace("True", "true")
+                        .replace("None", "null")
+                    )
+                    return json.loads(api_response)
+            return {}
+        else:
+            error_message = f"Expected status {expected_status}, but got {status_code}. Response: {response}"
+            raise Exception(error_message)
+
     """
     #############################
     xsoar related methods
@@ -89,11 +114,9 @@ def push_to_dataset(
             endpoint = urljoin(self.server_config.base_api_url, "logs/v1/event")
             additional_headers = {
                 "authorization": self.server_config.collector_token,
-                "content-type": (
-                    "application/json"
-                    if data_format.casefold == "json"
-                    else "text/plain"
-                ),
+                "content-type": "application/json"
+                if data_format.casefold == "json"
+                else "text/plain",
                 "content-encoding": "gzip",
             }
             token_type = "collector_token"
@@ -109,7 +132,8 @@ def push_to_dataset(
         )
         try:
             data = response.json()
-        except requests.exceptions.JSONDecodeError:  # type: ignore[attr-defined]
+        # type: ignore[attr-defined]
+        except requests.exceptions.JSONDecodeError:
             error = response.text
             err_msg = f"Failed to push using {token_type} - with status code {response.status_code}"
             err_msg += f"\n{error}" if error else ""
@@ -199,3 +223,114 @@ def get_ioc_rules(self):
             )
 
         return response
+
+    """
+    #############################
+    Alerts related methods
+    #############################
+    """
+
+    def create_alert_from_json(self, json_content: dict) -> int:
+        alert_payload = {"request_data": {"alert": json_content}}
+        endpoint = urljoin(
+            self.server_config.base_api_url, "/public_api/v1/alerts/create_alert"
+        )
+        res = self._xdr_client.post(endpoint, json=alert_payload)
+        alert_data = self._process_response(res.content, res.status_code, 200)
+        return alert_data["reply"]
+
+    def get_internal_alert_id(self, alert_external_id: str) -> int:
+        data = self.search_alerts(
+            filters=[
+                {
+                    "field": "external_id_list",
+                    "operator": "in",
+                    "value": [alert_external_id],
+                }
+            ]
+        )
+        return data["alerts"][0]["alert_id"]
+
+    def update_alert(self, alert_id: Union[str, list[str]], updated_data: dict) -> dict:
+        """
+        Args:
+            alert_id (str | list[str]): alert ids to edit.
+            updated_data (dict): The data to update the alerts with. https://cortex-panw.stoplight.io/docs/cortex-xsiam-1/rpt3p1ne2bwfe-update-alerts
+        """
+        alert_payload = {
+            "request_data": {"update_data": updated_data, "alert_id_list": alert_id}
+        }
+        endpoint = urljoin(
+            self.server_config.base_api_url, "/public_api/v1/alerts/update_alerts"
+        )
+        res = self._xdr_client.post(endpoint, json=alert_payload)
+        alert_data = self._process_response(res.content, res.status_code, 200)
+        return alert_data
+
+    def search_alerts(
+        self,
+        filters: list = None,
+        search_from: int = None,
+        search_to: int = None,
+        sort: dict = None,
+    ) -> dict:
+        """
+        filters should be a list of dicts contains field, operator, value.
+        For example:
+        [{field: alert_id_list, operator: in, value: [1,2,3,4]}]
+        Allowed values for fields - alert_id_list, alert_source, severity, creation_time
+        """
+        body = {
+            "request_data": {
+                "filters": filters,
+                "search_from": search_from,
+                "search_to": search_to,
+                "sort": sort,
+            }
+        }
+        endpoint = urljoin(
+            self.server_config.base_api_url, "/public_api/v1/alerts/get_alerts/"
+        )
+        res = self._xdr_client.post(endpoint, json=body)
+        return self._process_response(res.content, res.status_code, 200)["reply"]
+
+    def search_alerts_by_uuid(self, alert_uuids: list = None, filters: list = None):
+        if alert_uuids is None:
+            alert_uuids = []
+        alert_ids: list = []
+        res = self.search_alerts(filters=filters)
+        alerts: list = res.get("alerts")  # type: ignore
+        count: int = res.get("result_count")  # type: ignore
+
+        while len(alerts) > 0 and len(alert_uuids) > len(alert_ids):
+            for alert in alerts:
+                for uuid in alert_uuids:
+                    if alert.get("description").endswith(uuid):
+                        alert_ids.append(alert.get("alert_id"))
+
+            res = self.search_alerts(filters=filters, search_from=count)
+            alerts = res.get("alerts")  # type: ignore
+            count = res.get("result_count")  # type: ignore
+
+        return alert_ids
+
+    """
+    #############################
+    Playbooks related methods
+    #############################
+    """
+
+    def get_playbook_data(self, playbook_id: int) -> dict:
+        playbook_endpoint = f"/playbook/{playbook_id}"
+
+        response, status_code, _ = self._xsoar_client.generic_request(
+            playbook_endpoint, method="GET", accept="application/json"
+        )
+        return self._process_response(response, status_code, 200)
+
+    def update_playbook_input(self, playbook_id: str, new_inputs: dict):
+        saving_inputs_path = f"/playbook/inputs/{playbook_id}"
+        response, status_code, _ = self._xsoar_client.generic_request(
+            saving_inputs_path, method="POST", body={"inputs": new_inputs}
+        )
+        return self._process_response(response, status_code, 200)

demisto_sdk/commands/common/clients/xsoar_saas/xsoar_saas_api_client.py

@@ -42,6 +42,7 @@ def __init__(
                 "Content-Type": "application/json",
             }
         )
+        self.lcas_id = config.lcas_id
         super().__init__(
             config,
             client=client,

demisto_sdk/commands/common/constants.py

@@ -52,6 +52,7 @@
 XSIAM_TOKEN = "XSIAM_TOKEN"
 XSIAM_COLLECTOR_TOKEN = "XSIAM_COLLECTOR_TOKEN"
 DEMISTO_VERIFY_SSL = "DEMISTO_VERIFY_SSL"
+LCAS_ID = "LCAS_ID"
 
 # Logging
 DEMISTO_SDK_LOG_FILE_PATH = "DEMISTO_SDK_LOG_FILE_PATH"
@@ -2218,3 +2219,4 @@ class PlaybookTaskType(StrEnum):
 # Test types:
 TEST_PLAYBOOKS = "TestPlaybooks"
 TEST_MODELING_RULES = "TestModelingRules"
+Test_Use_Cases = "TestUseCases"