Add parser validation APIs and log-type CLI commands

api_module_mapping.md

@@ -296,6 +296,8 @@ Following shows mapping between SecOps [REST Resource](https://cloud.google.com/
 |logTypes.getLogTypeSetting                                                    |v1alpha|                                                            |                                       |
 |logTypes.legacySubmitParserExtension                                          |v1alpha|                                                            |                                       |
 |logTypes.list                                                                 |v1alpha|                                                            |                                       |
+|logTypes.getParserAnalysisReport                                              |v1alpha|chronicle.parser.get_analysis_report             |secops log-type get-analysis-report    |
+|logTypes.triggerGitHubChecks                                                  |v1alpha|chronicle.parser.trigger_github_checks           |secops log-type trigger-checks         |
 |logTypes.logs.export                                                          |v1alpha|                                                            |                                       |
 |logTypes.logs.get                                                             |v1alpha|                                                            |                                       |
 |logTypes.logs.import                                                          |v1alpha|chronicle.log_ingest.ingest_log                             |secops log ingest                      |

src/secops/chronicle/case.py

@@ -173,7 +173,7 @@ def execute_bulk_assign(
     Raises:
         APIError: If the API request fails
     """
-    body = {"casesIds": case_ids, "username": username}
+    body = {"casesIds": case_ids, "userName": username}
 
     return chronicle_request(
         client,

src/secops/chronicle/client.py

@@ -351,6 +351,10 @@
     create_watchlist as _create_watchlist,
     update_watchlist as _update_watchlist,
 )
+from secops.chronicle.parser import (
+    get_analysis_report as _get_analysis_report,
+    trigger_github_checks as _trigger_github_checks,
+)
 from secops.exceptions import SecOpsError
 
 
@@ -778,6 +782,51 @@ def update_watchlist(
             update_mask,
         )
 
+    def get_analysis_report(
+        self,
+        log_type: str,
+        parser_id: str,
+        report_id: str,
+    ) -> dict[str, Any]:
+        """Get a parser analysis report.
+        Args:
+            log_type: Log type of the parser.
+            parser_id: The ID of the parser.
+            report_id: The ID of the analysis report.
+        Returns:
+            Dictionary containing the analysis report.
+        Raises:
+            APIError: If the API request fails.
+        """
+        return _get_analysis_report(
+            self,
+            log_type=log_type,
+            parser_id=parser_id,
+            report_id=report_id,
+        )
+
+    def trigger_github_checks(
+        self,
+        associated_pr: str,
+        log_type: str,
+    ) -> dict[str, Any]:
+        """Trigger GitHub checks for a parser.
+
+        Args:
+            associated_pr: The PR string (e.g., "owner/repo/pull/123").
+            log_type: The string name of the LogType enum.
+
+        Returns:
+            Dictionary containing the response details.
+
+        Raises:
+            SecOpsError: If gRPC modules or client stub are not available.
+            APIError: If the gRPC API request fails.
+        """
+        return _trigger_github_checks(
+            self, associated_pr=associated_pr, log_type=log_type
+        )
+
     def get_stats(
         self,
         query: str,

src/secops/chronicle/parser.py

@@ -16,13 +16,16 @@
 
 import base64
 import json
+import logging
 from typing import Any
 
+from secops.chronicle.models import APIVersion
 from secops.chronicle.utils.format_utils import remove_none_values
 from secops.chronicle.utils.request_utils import (
     chronicle_paginated_request,
     chronicle_request,
 )
+from secops.exceptions import APIError, SecOpsError
 
 # Constants for size limits
 MAX_LOG_SIZE = 10 * 1024 * 1024  # 10MB per log
@@ -433,3 +436,121 @@ def run_parser(
                             print(f"Warning: Failed to parse statedump: {e}")
 
     return result
+
+
+def trigger_github_checks(
+    client: "ChronicleClient",
+    associated_pr: str,
+    log_type: str,
+    timeout: int = 60,
+) -> dict[str, Any]:
+    """Trigger GitHub checks for a parser.
+
+    Args:
+        client: ChronicleClient instance
+        associated_pr: The PR string (e.g., "owner/repo/pull/123").
+        log_type: The string name of the LogType enum.
+        timeout: Optional RPC timeout in seconds (default: 60).
+
+    Returns:
+        Dictionary containing the response details.
+
+    Raises:
+        SecOpsError: If input is invalid.
+        APIError: If the API request fails.
+    """
+
+    if not isinstance(log_type, str) or len(log_type.strip()) < 2:
+        raise SecOpsError("log_type must be a valid string of length >= 2")
+
+    if not isinstance(associated_pr, str) or not associated_pr.strip():
+        raise SecOpsError("associated_pr must be a non-empty string")
+
+    pr_parts = associated_pr.split("/")
+    if len(pr_parts) != 4 or pr_parts[2] != "pull" or not pr_parts[3].isdigit():
+        raise SecOpsError(
+            "associated_pr must be in the format 'owner/repo/pull/<number>'"
+        )
+    if not isinstance(timeout, int) or timeout < 0:
+        raise SecOpsError("timeout must be a non-negative integer")
+
+    try:
+        parsers = list_parsers(client, log_type=log_type)
+    except APIError as e:
+        raise APIError(
+            f"Failed to fetch parsers for log type {log_type}: {e}"
+        ) from e
+
+    if not parsers:
+        logging.info(
+            "No parsers found for log type %s. Using fallback parser ID.",
+            log_type,
+        )
+        parser_name = f"logTypes/{log_type}/parsers/-"
+    else:
+        if len(parsers) > 1:
+            logging.warning(
+                "Multiple parsers found for log type %s. Using the first one.",
+                log_type,
+            )
+        parser_name = parsers[0]["name"]
+
+    endpoint_path = f"{parser_name}:runAnalysis"
+    payload = {
+        "report_type": "GITHUB_PARSER_VALIDATION",
+        "pull_request": associated_pr,
+    }
+
+    return chronicle_request(
+        client=client,
+        method="POST",
+        api_version="v1alpha",
+        endpoint_path=endpoint_path,
+        json=payload,
+        timeout=timeout,
+    )
+
+
+def get_analysis_report(
+    client: "ChronicleClient",
+    log_type: str,
+    parser_id: str,
+    report_id: str,
+    timeout: int = 60,
+) -> dict[str, Any]:
+    """Get a parser analysis report.
+
+    Args:
+        client: ChronicleClient instance
+        log_type: Log type of the parser.
+        parser_id: The ID of the parser.
+        report_id: The ID of the analysis report.
+        timeout: Optional timeout in seconds (default: 60).
+
+    Returns:
+        Dictionary containing the analysis report.
+
+    Raises:
+        SecOpsError: If input is invalid.
+        APIError: If the API request fails.
+    """
+    if not isinstance(log_type, str) or not log_type.strip():
+        raise SecOpsError("log_type must be a non-empty string")
+    if not isinstance(parser_id, str) or not parser_id.strip():
+        raise SecOpsError("parser_id must be a non-empty string")
+    if not isinstance(report_id, str) or not report_id.strip():
+        raise SecOpsError("report_id must be a non-empty string")
+    if not isinstance(timeout, int) or timeout < 0:
+        raise SecOpsError("timeout must be a non-negative integer")
+
+    endpoint_path = (
+        f"logTypes/{log_type}/parsers/{parser_id}/analysisReports/{report_id}"
+    )
+
+    return chronicle_request(
+        client=client,
+        method="GET",
+        api_version=APIVersion.V1ALPHA,
+        endpoint_path=endpoint_path,
+        timeout=timeout,
+    )

src/secops/cli/cli_client.py

@@ -26,6 +26,7 @@
 from secops.cli.commands.investigation import setup_investigation_command
 from secops.cli.commands.iocs import setup_iocs_command
 from secops.cli.commands.log import setup_log_command
+from secops.cli.commands.log_type import setup_log_type_commands
 from secops.cli.commands.log_processing import (
     setup_log_processing_command,
 )
@@ -168,6 +169,7 @@ def build_parser() -> argparse.ArgumentParser:
     setup_investigation_command(subparsers)
     setup_iocs_command(subparsers)
     setup_log_command(subparsers)
+    setup_log_type_commands(subparsers)
     setup_log_processing_command(subparsers)
     setup_parser_command(subparsers)
     setup_parser_extension_command(subparsers)

src/secops/cli/commands/log_type.py

@@ -0,0 +1,124 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""CLI for ParserValidationToolingService under Log Type command group"""
+
+import sys
+
+from secops.cli.utils.formatters import output_formatter
+from secops.exceptions import APIError, SecOpsError
+
+
+def setup_log_type_commands(subparsers):
+    """Set up the log_type service commands for Parser Validation."""
+    log_type_parser = subparsers.add_parser(
+        "log-type",
+        help="Log Type related operations (including Parser Validation)",
+    )
+
+    log_type_subparsers = log_type_parser.add_subparsers(
+        title="Log Type Commands",
+        dest="log_type_command",
+        help="Log Type sub-command to execute",
+    )
+
+    if sys.version_info >= (3, 7):
+        log_type_subparsers.required = True
+
+    log_type_parser.set_defaults(
+        func=lambda args, chronicle: log_type_parser.print_help()
+    )
+
+    # --- trigger-checks command ---
+    trigger_github_checks_parser = log_type_subparsers.add_parser(
+        "trigger-checks", help="Trigger GitHub checks for a parser"
+    )
+    trigger_github_checks_parser.add_argument(
+        "--associated-pr",
+        "--associated_pr",
+        required=True,
+        help='The PR string (e.g., "owner/repo/pull/123").',
+    )
+    trigger_github_checks_parser.add_argument(
+        "--log-type",
+        "--log_type",
+        required=True,
+        help='The string name of the LogType enum (e.g., "DUMMY_LOGTYPE").',
+    )
+    trigger_github_checks_parser.set_defaults(
+        func=handle_trigger_checks_command
+    )
+
+    # --- get-analysis-report command ---
+    get_report_parser = log_type_subparsers.add_parser(
+        "get-analysis-report", help="Get a parser analysis report"
+    )
+    get_report_parser.add_argument(
+        "--log-type",
+        "--log_type",
+        required=True,
+        help="The log type of the parser.",
+    )
+    get_report_parser.add_argument(
+        "--parser-id",
+        "--parser_id",
+        required=True,
+        help="The ID of the parser.",
+    )
+    get_report_parser.add_argument(
+        "--report-id",
+        "--report_id",
+        required=True,
+        help="The ID of the analysis report.",
+    )
+    get_report_parser.set_defaults(func=handle_get_analysis_report_command)
+
+
+def handle_trigger_checks_command(args, chronicle):
+    """Handle trigger checks command."""
+    try:
+        result = chronicle.trigger_github_checks(
+            associated_pr=args.associated_pr,
+            log_type=args.log_type,
+        )
+        output_formatter(result, args.output)
+    except APIError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+    except SecOpsError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+    except Exception as e:  # pylint: disable=broad-exception-caught
+        print(f"Error triggering GitHub checks: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
+def handle_get_analysis_report_command(args, chronicle):
+    """Handle get analysis report command."""
+    try:
+        result = chronicle.get_analysis_report(
+            log_type=args.log_type,
+            parser_id=args.parser_id,
+            report_id=args.report_id,
+        )
+        output_formatter(result, args.output)
+    except APIError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+    except SecOpsError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+    except Exception as e:  # pylint: disable=broad-exception-caught
+        print(f"Error fetching analysis report: {e}", file=sys.stderr)
+        sys.exit(1)

src/secops/cli/commands/rule.py

@@ -14,7 +14,6 @@
 #
 """Google SecOps CLI rule commands"""
 
-from ast import arg
 import json
 import sys
 

tests/chronicle/test_case.py

@@ -133,7 +133,7 @@ def test_execute_bulk_assign_success(chronicle_client):
         assert call_args[1]["endpoint_path"] == "cases:executeBulkAssign"
         assert call_args[1]["json"] == {
             "casesIds": [123, 456],
-            "username": "user@example.com",
+            "userName": "user@example.com",
         }
         assert result == {}