Linting

Author: eric-forte-elastic

detection_rules/rule.py

@@ -1470,7 +1470,7 @@ def get_packaged_integrations(
         # if both exist, rule tags are only used if defined in definitions for non-dataset packages
         # of machine learning analytic packages
 
-        rule_integrations = meta.get("integration") or []
+        rule_integrations: str | list[str] = meta.get("integration") or []
         for integration in rule_integrations:
             ineligible_integrations = [
                 *definitions.NON_DATASET_PACKAGES,

detection_rules/rule_validators.py

@@ -801,12 +801,13 @@ def log(val: str) -> None:
                 print(f"{rule_id}:", val)
 
         stack_version = ""
-        kibana_details = kibana_client.get("/api/status")
+        kibana_details: dict[str, Any] = kibana_client.get("/api/status", {})  # type: ignore[reportUnknownVariableType]
+        if "version" not in kibana_details:
+            raise ValueError("Failed to retrieve Kibana details.")
         stack_version = str(kibana_details["version"]["number"])
-
         log(f"Validating against {stack_version} stack")
 
-        indices_str, indices = utils.get_esql_query_indices(contents.data.query)
+        indices_str, indices = utils.get_esql_query_indices(contents.data.query)  # type: ignore[reportUnknownVariableType]
         log(f"Extracted indices from query: {', '.join(indices)}")
 
         # Get mappings for all matching existing index templates
@@ -820,10 +821,10 @@ def log(val: str) -> None:
         full_index_str = self.create_remote_indices(elastic_client, existing_mappings, index_lookup, log)
 
         # Replace all sources with the test indices
-        query = contents.data.query
-        query = query.replace(indices_str, full_index_str)
+        query = contents.data.query  # type: ignore[reportUnknownVariableType]
+        query = query.replace(indices_str, full_index_str)  # type: ignore[reportUnknownVariableType]
 
-        query_columns = self.execute_query_against_indices(elastic_client, query, full_index_str, log)
+        query_columns = self.execute_query_against_indices(elastic_client, query, full_index_str, log)  # type: ignore[reportUnknownVariableType]
 
         # Validation all fields (columns) are either dynamic fields or correctly mapped
         # against the combined mapping of all the indices

detection_rules/utils.py

@@ -553,18 +553,18 @@ def convert_to_nested_schema(flat_schemas: dict[str, str]) -> dict[str, Any]:
         current_level = nested_schema
 
         for part in parts[:-1]:
-            current_level = current_level.setdefault(part, {}).setdefault("properties", {})
+            current_level = current_level.setdefault(part, {}).setdefault("properties", {})  # type: ignore[reportUnknownVariableType]
 
         current_level[parts[-1]] = {"type": value}
 
-    return nested_schema
+    return nested_schema  # type: ignore[reportUnknownVariableType]
 
 
 def combine_dicts(dest: dict[Any, Any], src: dict[Any, Any]) -> None:
     """Combine two dictionaries recursively."""
     for k, v in src.items():
         if k in dest and isinstance(dest[k], dict) and isinstance(v, dict):
-            combine_dicts(dest[k], v)
+            combine_dicts(dest[k], v)  # type: ignore[reportUnknownVariableType]
         else:
             dest[k] = v
 
@@ -585,12 +585,12 @@ def flat_schema_to_index_mapping(flat_schema: dict[str, str]) -> dict[str, Any]:
         current_level = result
 
         for part in parts[:-1]:
-            node = current_level.setdefault(part, {})
+            node = current_level.setdefault(part, {})  # type: ignore[reportUnknownVariableType]
 
             if "type" in node and node["type"] not in ("nested", "object"):
-                current_level = node.setdefault("fields", {})
+                current_level = node.setdefault("fields", {})  # type: ignore[reportUnknownVariableType]
             else:
-                current_level = node.setdefault("properties", {})
+                current_level = node.setdefault("properties", {})  # type: ignore[reportUnknownVariableType]
 
         leaf_key = parts[-1]
         current_level[leaf_key] = {"type": field_type}
@@ -604,16 +604,16 @@ def flat_schema_to_index_mapping(flat_schema: dict[str, str]) -> dict[str, Any]:
         if field_type == "alias":
             current_level[leaf_key]["path"] = "@timestamp"
 
-    return result
+    return result  # type: ignore[reportUnknownVariableType]
 
 
 def get_column_from_index_mapping_schema(keys: list[str], current_schema: dict[str, Any] | None) -> str | None:
     """Recursively traverse the schema to find the type of the column."""
     key = keys[0]
     if not current_schema:
         return None
-    column = current_schema.get(key) or {}
-    column_type = column.get("type") if column else None
+    column = current_schema.get(key) or {}  # type: ignore[reportUnknownVariableType]
+    column_type = column.get("type") if column else None  # type: ignore[reportUnknownVariableType]
     if not column_type and len(keys) > 1:
-        return get_column_from_index_mapping_schema(keys[1:], current_schema=column.get("properties"))
-    return column_type
+        return get_column_from_index_mapping_schema(keys[1:], current_schema=column.get("properties"))  # type: ignore[reportUnknownVariableType]
+    return column_type  # type: ignore[reportUnknownVariableType]

tests/test_rules_remote.py

@@ -5,9 +5,10 @@
 
 import unittest
 
+from elasticsearch import BadRequestError
+
 from detection_rules.misc import get_default_config, get_elasticsearch_client, get_kibana_client, getdefault
 from detection_rules.rule_validators import ESQLValidator
-from elasticsearch import BadRequestError
 
 from .base import BaseRuleTest