Add parser validation APIs and log-type CLI commands

CHANGELOG.md

@@ -5,6 +5,15 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.40.0] - 2026-04-06
+### Added
+- Parser validation methods
+  - `trigger_github_checks()` - Trigger GitHub checks for a parser against an associated pull request
+  - `get_analysis_report()` - Retrieve a completed parser analysis report
+- CLI support for parser validation commands
+  - `secops log-type trigger-checks` - Trigger parser validation checks for a PR
+  - `secops log-type get-analysis-report` - Get details of a specific analysis report
+
 ## [0.39.0] - 2026-04-02
 ### Updated
 - Refactored Chronicle modules to use centralized `chronicle_request` and `chronicle_paginated_request` helper functions for improved code consistency and maintainability

CLI.md

@@ -696,6 +696,22 @@ Error messages are detailed and help identify issues:
 - Size limit violations
 - API-specific errors
 
+#### Parser Validation
+
+You can trigger and retrieve analysis reports for parsers associated with GitHub pull requests.
+
+Trigger GitHub checks for a parser:
+
+```bash
+secops log-type trigger-checks --log-type "WINDOWS_AD" --associated-pr "owner/repo/pull/123"
+```
+
+Get a parser analysis report:
+
+```bash
+secops log-type get-analysis-report --log-type "WINDOWS_AD" --parser-id "pa_12345" --report-id "report_12345"
+```
+
 ### Parser Extension Management
 
 Parser extensions provide a flexible way to extend the capabilities of existing default (or custom) parsers without replacing them.

README.md

@@ -1898,6 +1898,27 @@ This workflow is useful for:
 - Re-processing logs with updated parsers
 - Debugging parsing issues
 
+### Parser Validation
+
+Trigger and retrieve analysis reports for parsers associated with GitHub pull requests:
+
+```python
+# Trigger GitHub checks for a parser against a PR
+response = chronicle.trigger_github_checks(
+    associated_pr="owner/repo/pull/123",
+    log_type="WINDOWS_AD"
+)
+print(f"Triggered checks: {response}")
+
+# Retrieve the analysis report
+report = chronicle.get_analysis_report(
+    log_type="WINDOWS_AD",
+    parser_id="pa_1234567890",
+    report_id="report_0987654321"
+)
+print(f"Analysis report: {report}")
+```
+
 ## Parser Extension
 
 Parser extensions provide a flexible way to extend the capabilities of existing default (or custom) parsers without replacing them. The extensions let you customize the parser pipeline by adding new parsing logic, extracting and transforming fields, and updating or removing UDM field mappings.

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                      |

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "secops"
-version = "0.39.0"
+version = "0.40.0"
 description = "Python SDK for wrapping the Google SecOps API for common use cases"
 readme = "README.md"
 requires-python = ">=3.10"

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,59 @@ def update_watchlist(
             update_mask,
         )
 
+    def get_analysis_report(
+        self,
+        log_type: str,
+        parser_id: str,
+        report_id: str,
+        timeout: int = 60,
+    ) -> 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.
+            timeout: Optional timeout in seconds (default: 60).
+        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,
+            timeout=timeout,
+        )
+
+    def trigger_github_checks(
+        self,
+        associated_pr: str,
+        log_type: str,
+        timeout: int = 60,
+    ) -> 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.
+            timeout: Optional request timeout in seconds (default: 60).
+
+        Returns:
+            Dictionary containing the response details.
+
+        Raises:
+            SecOpsError: If modules or client stub are not available.
+            APIError: If the API request fails.
+        """
+        return _trigger_github_checks(
+            self,
+            associated_pr=associated_pr,
+            log_type=log_type,
+            timeout=timeout,
+        )
+
     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 request 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)