add recording management command to azpysdk

eng/tools/azure-sdk-tools/azpysdk/main.py

@@ -40,6 +40,8 @@
 from .samples import samples
 from .devtest import devtest
 from .optional import optional
+from .update_snippets import update_snippets
+from .recording import recording
 
 from ci_tools.logging import configure_logging, logger
 
@@ -110,6 +112,8 @@ def build_parser() -> argparse.ArgumentParser:
     samples().register(subparsers, [common])
     devtest().register(subparsers, [common])
     optional().register(subparsers, [common])
+    update_snippets().register(subparsers, [common])
+    recording().register(subparsers, [])
 
     return parser
 

eng/tools/azure-sdk-tools/azpysdk/recording.py

@@ -0,0 +1,171 @@
+"""Manage test recording assets with the test-proxy tool.
+
+This command provides a unified interface for managing test recordings
+stored in the azure-sdk-assets repository.
+"""
+
+import argparse
+import os
+import sys
+import shlex
+import subprocess
+from pathlib import Path
+from typing import Optional, List
+
+from ci_tools.variables import discover_repo_root
+from ci_tools.logging import logger
+
+from .Check import Check
+
+
+class recording(Check):
+    """Manage test recording assets via the test-proxy tool.
+
+    This command wraps the test-proxy asset sync functionality, providing
+    verbs for pushing, restoring, resetting, locating, and showing recording
+    assets.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+
+    def register(
+        self, subparsers: "argparse._SubParsersAction", parent_parsers: Optional[List[argparse.ArgumentParser]] = None
+    ) -> None:
+        """Register the `recording` command with subcommands for asset management."""
+        parents = parent_parsers or []
+
+        # Create main recording parser
+        p = subparsers.add_parser(
+            "recording",
+            parents=parents,
+            help="Manage test recording assets",
+            description="Manage test recording assets stored in azure-sdk-assets repository",
+        )
+
+        # Create subparsers for recording verbs
+        recording_subparsers = p.add_subparsers(
+            title="recording commands",
+            dest="recording_verb",
+            required=True,
+            help="Asset management operations",
+        )
+
+        # Common argument for assets.json path
+        path_arg = argparse.ArgumentParser(add_help=False)
+        path_arg.add_argument(
+            "-p",
+            "--path",
+            default="assets.json",
+            help='The relative path to your package\'s assets.json file. Default is "assets.json" in current directory.',
+        )
+
+        # Push command
+        push_parser = recording_subparsers.add_parser(
+            "push",
+            parents=[path_arg],
+            help="Push recording updates to a new assets repo tag and update assets.json",
+        )
+        push_parser.set_defaults(func=self.run)
+
+        # Restore command
+        restore_parser = recording_subparsers.add_parser(
+            "restore",
+            parents=[path_arg],
+            help="Fetch recordings from the assets repo based on the tag in assets.json",
+        )
+        restore_parser.set_defaults(func=self.run)
+
+        # Reset command
+        reset_parser = recording_subparsers.add_parser(
+            "reset",
+            parents=[path_arg],
+            help="Discard any pending changes to recordings based on the tag in assets.json",
+        )
+        reset_parser.set_defaults(func=self.run)
+
+        # Locate command
+        locate_parser = recording_subparsers.add_parser(
+            "locate",
+            parents=[path_arg],
+            help="Print the location of the library's locally cached recordings",
+        )
+        locate_parser.set_defaults(func=self.run)
+
+        # Show command
+        show_parser = recording_subparsers.add_parser(
+            "show",
+            parents=[path_arg],
+            help="Print the contents of the provided assets.json file",
+        )
+        show_parser.set_defaults(func=self.run)
+
+    def run(self, args: argparse.Namespace) -> int:
+        """Execute the recording management command."""
+
+        if not hasattr(args, 'recording_verb') or not args.recording_verb:
+            logger.error("No recording verb specified. Use one of: push, restore, reset, locate, show")
+            return 1
+
+        verb = args.recording_verb
+        assets_path = args.path
+
+        logger.info(f"Running 'azpysdk recording {verb}' with assets.json path: {assets_path}")
+
+        # Normalize path for cross-platform compatibility
+        normalized_path = assets_path.replace("\\", "/")
+
+        # Get repository root
+        try:
+            repo_root = self._ascend_to_root(os.getcwd())
+            logger.info(f"Repository root: {repo_root}")
+        except Exception as e:
+            logger.error(f"Failed to find repository root: {e}")
+            return 1
+
+        # Prepare test-proxy tool
+        try:
+            tool_name = self._prepare_local_tool(repo_root)
+            logger.info(f"Using test-proxy tool: {tool_name}")
+        except Exception as e:
+            logger.error(f"Failed to prepare test-proxy tool: {e}")
+            return 1
+
+        # Build command based on verb
+        config_commands = {"locate", "show"}
+        if verb in config_commands:
+            command = f"{tool_name} config {verb.lower()} -a {normalized_path}"
+        else:
+            command = f"{tool_name} {verb.lower()} -a {normalized_path}"
+
+        logger.info(f"Executing command: {command}")
+
+        # Execute the command
+        try:
+            result = subprocess.run(
+                shlex.split(command),
+                stdout=sys.stdout,
+                stderr=sys.stderr,
+                check=False,
+            )
+
+            if result.returncode != 0:
+                logger.error(f"Command failed with exit code {result.returncode}")
+                return result.returncode
+
+            logger.info(f"Successfully completed '{verb}' operation")
+            return 0
+
+        except Exception as e:
+            logger.error(f"Failed to execute command: {e}")
+            return 1
+
+    def _ascend_to_root(self, start_dir: str) -> str:
+        """Ascend from start directory until finding a .git folder."""
+        from devtools_testutils.proxy_startup import ascend_to_root
+        return ascend_to_root(start_dir)
+
+    def _prepare_local_tool(self, repo_root: str) -> str:
+        """Prepare and return the path to the test-proxy executable."""
+        from devtools_testutils.proxy_startup import prepare_local_tool
+        return prepare_local_tool(repo_root)

scripts/manage_recordings.py

@@ -4,24 +4,63 @@
 # license information.
 # --------------------------------------------------------------------------
 
+# ============================================================================
+# DEPRECATION NOTICE
+# ============================================================================
+# This script is deprecated and maintained only for backward compatibility.
+# 
+# Please use the 'azpysdk recording' command instead:
+#
+#   azpysdk recording push -p sdk/keyvault/azure-keyvault-keys/assets.json
+#   azpysdk recording restore
+#   azpysdk recording reset
+#   azpysdk recording locate
+#   azpysdk recording show
+#
+# For more information, run: azpysdk recording --help
+# ============================================================================
+
 import argparse
 import os
 import shlex
 import subprocess
 import sys
+import warnings
 
 from devtools_testutils.proxy_startup import (
     ascend_to_root,
     prepare_local_tool,
 )
 
+# Emit deprecation warning
+warnings.warn(
+    "\n"
+    "=" * 80 + "\n"
+    "DEPRECATION WARNING: This script is deprecated.\n"
+    "Please use 'azpysdk recording' command instead.\n"
+    "\n"
+    "Examples:\n"
+    "  azpysdk recording push -p sdk/keyvault/azure-keyvault-keys/assets.json\n"
+    "  azpysdk recording restore\n"
+    "  azpysdk recording reset\n"
+    "\n"
+    "Run 'azpysdk recording --help' for more information.\n"
+    "=" * 80,
+    DeprecationWarning,
+    stacklevel=2,
+)
+
 
 TOOL_ENV_VAR = "PROXY_PID"
 
 
 # This file contains a script for managing test recordings in the azure-sdk-assets repository.
 #
-# INSTRUCTIONS FOR USE:
+# ============================================================================
+# DEPRECATION NOTICE: This script is deprecated. Use 'azpysdk recording' instead.
+# ============================================================================
+#
+# LEGACY INSTRUCTIONS FOR USE (for backward compatibility only):
 #
 # - Set your working directory to be inside your local copy of the azure-sdk-for-python repository.
 # - Run the following command:
@@ -33,11 +72,17 @@
 #
 #     `python scripts/manage_recordings.py push -p sdk/keyvault/azure-keyvault-keys/assets.json`
 #
+#   NEW RECOMMENDED COMMAND:
+#     `azpysdk recording push -p sdk/keyvault/azure-keyvault-keys/assets.json`
+#
 #   If this script is run from the directory containing an assets.json file, no path needs to be provided. For example,
 #   with a working directory at the azure-keyvault-keys package root:
 #
 #     `python ../../../scripts/manage_recordings.py push`
 #
+#   NEW RECOMMENDED COMMAND:
+#     `azpysdk recording push`
+#
 # - In addition to "push", you can also use the "restore" or "reset" verbs in the same command format.
 #
 #   * locate: prints the location of the library's locally cached recordings.