mattermost/python-sdk/scripts/audit_client_coverage.py

#!/usr/bin/env python3
# Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
# See LICENSE.txt for license information.

"""
Audit script for verifying that all gRPC RPCs have corresponding client methods.

This script inspects the PluginAPIClient class and compares its methods against
the RPCs defined in the protobuf service definition. It reports any RPCs that
are missing Python client method implementations.

Usage:
    python scripts/audit_client_coverage.py
    python scripts/audit_client_coverage.py --include '(User|Team|Channel)'
    python scripts/audit_client_coverage.py --exclude '(Post|File|KV)'
    python scripts/audit_client_coverage.py --include '(User|Team|Channel)' --exclude '(Post|File|KV)'

The script:
1. Loads the generated protobuf service descriptor
2. Extracts all RPC method names
3. Applies include/exclude regex filters
4. Inspects PluginAPIClient for corresponding methods
5. Reports coverage statistics and any missing methods
"""

import argparse
import inspect
import re
import sys
from typing import List, Set, Tuple


def camel_to_snake(name: str) -> str:
    """
    Convert CamelCase to snake_case.

    Examples:
        GetUser -> get_user
        GetUserByEmail -> get_user_by_email
        HasPermissionToChannel -> has_permission_to_channel
    """
    # Insert underscore before uppercase letters and convert to lower
    s1 = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", name)
    return re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1).lower()


def get_rpc_names_from_stub() -> List[str]:
    """
    Get all RPC names from the PluginAPI gRPC stub.

    We inspect the stub class attributes to find all registered RPC methods.
    """
    try:
        from mattermost_plugin.grpc import api_pb2_grpc
    except ImportError:
        print("ERROR: Could not import api_pb2_grpc. Are protos generated?")
        sys.exit(1)

    # Get all attributes of the stub class
    stub_class = api_pb2_grpc.PluginAPIStub

    # Find RPC methods by looking at what the __init__ sets up
    # We can infer RPC names from the stub source or by introspection
    rpc_names = []

    # The stub's __init__ signature shows all the RPCs it configures
    # We'll look at the source to extract RPC names
    source = inspect.getsource(stub_class.__init__)

    # Find patterns like "self.CreateUser = channel.unary_unary("
    pattern = r"self\.(\w+)\s*=\s*channel\.unary_unary\("
    matches = re.findall(pattern, source)

    rpc_names.extend(matches)

    # Also check for streaming patterns if any
    pattern = r"self\.(\w+)\s*=\s*channel\.(?:unary_stream|stream_unary|stream_stream)\("
    matches = re.findall(pattern, source)
    rpc_names.extend(matches)

    return sorted(set(rpc_names))


def get_client_methods() -> Set[str]:
    """
    Get all public methods from PluginAPIClient.
    """
    try:
        from mattermost_plugin.client import PluginAPIClient
    except ImportError:
        print("ERROR: Could not import PluginAPIClient. Is the SDK installed?")
        sys.exit(1)

    methods = set()

    for name, value in inspect.getmembers(PluginAPIClient, predicate=inspect.isfunction):
        # Skip private methods
        if name.startswith("_"):
            continue
        methods.add(name)

    return methods


def filter_rpcs(
    rpc_names: List[str],
    include_pattern: str = "",
    exclude_pattern: str = "",
) -> List[str]:
    """
    Filter RPC names based on include/exclude patterns.
    """
    filtered = rpc_names

    if include_pattern:
        include_re = re.compile(include_pattern)
        filtered = [n for n in filtered if include_re.search(n)]

    if exclude_pattern:
        exclude_re = re.compile(exclude_pattern)
        filtered = [n for n in filtered if not exclude_re.search(n)]

    return filtered


def find_missing_and_extra(
    rpc_names: List[str],
    client_methods: Set[str],
) -> Tuple[List[str], List[str]]:
    """
    Find RPCs that are missing client methods and vice versa.

    Returns:
        Tuple of (missing_rpcs, extra_client_methods)
        - missing_rpcs: RPCs without corresponding client methods
        - extra_client_methods: Client methods that don't map to an RPC in scope
    """
    missing = []
    expected_methods = set()

    for rpc_name in rpc_names:
        expected_method = camel_to_snake(rpc_name)
        expected_methods.add(expected_method)

        if expected_method not in client_methods:
            missing.append(rpc_name)

    # Find extra methods in the scope that aren't RPCs
    # (This is informational - extra methods are fine)

    return missing, []


def main():
    parser = argparse.ArgumentParser(
        description="Audit PluginAPIClient coverage against gRPC service definition",
    )
    parser.add_argument(
        "--include",
        default="",
        help="Regex pattern to include RPC names (applied first)",
    )
    parser.add_argument(
        "--exclude",
        default="",
        help="Regex pattern to exclude RPC names (applied after include)",
    )
    parser.add_argument(
        "--verbose", "-v",
        action="store_true",
        help="Show detailed output including all RPCs",
    )

    args = parser.parse_args()

    # Get all RPC names
    all_rpcs = get_rpc_names_from_stub()
    print(f"Total RPCs in service: {len(all_rpcs)}")

    # Filter RPCs
    filtered_rpcs = filter_rpcs(all_rpcs, args.include, args.exclude)
    print(f"RPCs after filtering: {len(filtered_rpcs)}")

    if args.include:
        print(f"  Include pattern: {args.include}")
    if args.exclude:
        print(f"  Exclude pattern: {args.exclude}")

    # Get client methods
    client_methods = get_client_methods()
    print(f"Client methods: {len(client_methods)}")

    # Find missing
    missing, _ = find_missing_and_extra(filtered_rpcs, client_methods)

    if args.verbose:
        print("\nFiltered RPCs:")
        for rpc in filtered_rpcs:
            method = camel_to_snake(rpc)
            status = "OK" if method in client_methods else "MISSING"
            print(f"  {rpc} -> {method} [{status}]")

    print()

    # Report results
    covered = len(filtered_rpcs) - len(missing)
    percentage = (covered / len(filtered_rpcs) * 100) if filtered_rpcs else 100

    print(f"Coverage: {covered}/{len(filtered_rpcs)} ({percentage:.1f}%)")

    if missing:
        print(f"\nMissing client methods for {len(missing)} RPCs:")
        for rpc in missing:
            method = camel_to_snake(rpc)
            print(f"  - {rpc} (expected method: {method})")

        return 1
    else:
        print("\nAll in-scope RPCs have corresponding client methods!")
        return 0


if __name__ == "__main__":
    sys.exit(main())