[Resource] az bicep: Add snapshot and run subcommands (#33398)

Author: shenglol

src/azure-cli/azure/cli/command_modules/resource/_help.py

@@ -2763,6 +2763,42 @@
     text: az bicep lint --file {bicep_file} --diagnostics-format {diagnostics_format}
 """
 
+helps['bicep snapshot'] = """
+type: command
+short-summary: Capture or validate a snapshot of the resources predicted to be deployed by a .bicepparam file.
+long-summary: |
+    Compiles a .bicepparam file together with its referenced Bicep template and writes a deployment
+    snapshot (a `*.snapshot.json` file) next to the .bicepparam file. When run with `--mode Validate`,
+    the existing snapshot is compared against the current template and the command fails if they
+    differ. This command requires Bicep CLI v0.41.2 or later.
+examples:
+  - name: Capture a snapshot for a .bicepparam file.
+    text: az bicep snapshot --file main.bicepparam
+  - name: Validate that the existing snapshot still matches the current template.
+    text: az bicep snapshot --file main.bicepparam --mode Validate
+  - name: Capture a snapshot with explicit Azure context.
+    text: az bicep snapshot --file main.bicepparam --subscription-id 00000000-0000-0000-0000-000000000000 --resource-group myRg --location westus
+"""
+
+helps['bicep run'] = """
+type: command
+short-summary: Forward a raw command to the installed Bicep CLI.
+long-summary: |
+    Runs the Bicep CLI with the arguments supplied via `--command`, allowing use of Bicep CLI
+    features that do not yet have a dedicated `az bicep` wrapper. The string passed to
+    `--command` is split using shell-style quoting and forwarded to the Bicep CLI verbatim.
+    When the forwarded command itself starts with `--` (for example `--version`), use the
+    `--command=<value>` form so the CLI parser does not mistake the value for another option.
+
+    Because the value is forwarded to the Bicep CLI without validation, do not pass strings
+    derived from untrusted input.
+examples:
+  - name: Forward a build command to the Bicep CLI.
+    text: az bicep run --command "build main.bicep"
+  - name: Show the Bicep CLI help (use the --command=<value> form for option-like values).
+    text: az bicep run --command=--help
+"""
+
 helps['resourcemanagement'] = """
 type: group
 short-summary: resourcemanagement CLI command group.

src/azure-cli/azure/cli/command_modules/resource/_params.py

@@ -757,6 +757,27 @@ def load_arguments(self, _):
         c.argument('no_restore', arg_type=bicep_no_restore_type, help="When set, generates the parameters file without restoring external modules.")
         c.argument('diagnostics_format', arg_type=get_enum_type(['default', 'sarif']), help="Set diagnostics format.")
 
+    with self.argument_context('bicep snapshot') as c:
+        c.argument('file', arg_type=bicep_file_type, help="The path to the .bicepparam file to capture a snapshot for.")
+        c.argument('mode', arg_type=get_enum_type(['Overwrite', 'Validate']),
+                   help="The snapshot mode. 'Overwrite' (default) writes the snapshot file. 'Validate' compares the existing snapshot against the current template and fails if differences are detected.")
+        c.argument('tenant_id', options_list=['--tenant-id'],
+                   help="Tenant ID forwarded to the Bicep CLI as the deployment context used to resolve `existing` references when capturing the snapshot. This does not affect Azure CLI authentication.")
+        c.argument('subscription_id', options_list=['--subscription-id'],
+                   help="Subscription ID forwarded to the Bicep CLI as the deployment context used to resolve `existing` references when capturing the snapshot. This does not affect Azure CLI authentication; use the global `--subscription` argument to switch the active subscription.")
+        c.argument('management_group_id', options_list=['--management-group-id'],
+                   help="Management group ID forwarded to the Bicep CLI as the deployment context used to resolve `existing` references when capturing the snapshot.")
+        c.argument('location', arg_type=get_location_type(self.cli_ctx),
+                   help="Location forwarded to the Bicep CLI as the deployment context used to resolve `existing` references when capturing the snapshot.")
+        c.argument('resource_group', arg_type=resource_group_name_type,
+                   help="Resource group name forwarded to the Bicep CLI as the deployment context used to resolve `existing` references when capturing the snapshot.")
+        c.argument('deployment_name', options_list=['--deployment-name'],
+                   help="Deployment name forwarded to the Bicep CLI as the deployment context used to resolve `existing` references when capturing the snapshot.")
+
+    with self.argument_context('bicep run') as c:
+        c.argument('command_string', options_list=['--command', '-c'],
+                   help="The Bicep CLI command to run, including its arguments, as a single quoted string (e.g. \"build main.bicep\").")
+
     with self.argument_context('resourcemanagement private-link create') as c:
         c.argument('resource_group', arg_type=resource_group_name_type,
                    help='The name of the resource group.')

src/azure-cli/azure/cli/command_modules/resource/commands.py

@@ -638,6 +638,8 @@ def load_command_table(self, _):
         g.custom_command('list-versions', 'list_bicep_cli_versions')
         g.custom_command('generate-params', 'generate_params_file')
         g.custom_command('lint', 'lint_bicep_file')
+        g.custom_command('snapshot', 'snapshot_bicep_file')
+        g.custom_command('run', 'run_bicep_cli_passthrough')
 
     with self.command_group('resourcemanagement private-link', resource_resourcemanagementprivatelink_sdk, resource_type=ResourceType.MGMT_RESOURCE_PRIVATELINKS) as g:
         g.custom_command('create', 'create_resourcemanager_privatelink')

src/azure-cli/azure/cli/command_modules/resource/custom.py

@@ -23,7 +23,7 @@
 from azure.mgmt.resource.deployments.models import DeploymentMode
 import azure.mgmt.resource.deploymentstacks.models as StackModels
 
-from azure.cli.core.azclierror import ArgumentUsageError, InvalidArgumentValueError, ResourceNotFoundError
+from azure.cli.core.azclierror import ArgumentUsageError, InvalidArgumentValueError, ResourceNotFoundError, ValidationError
 from azure.cli.core.parser import IncorrectUsageError
 from azure.cli.core.util import get_file_json, read_file_content, shell_safe_json_parse, sdk_no_wait
 from azure.cli.core.commands import LongRunningOperation
@@ -4580,6 +4580,63 @@ def lint_bicep_file(cmd, file, no_restore=None, diagnostics_format=None):
         logger.error("az bicep lint could not be executed with the current version of Bicep CLI. Please upgrade Bicep CLI to v%s or later.", minimum_supported_version)
 
 
+def snapshot_bicep_file(cmd, file, mode=None, tenant_id=None, subscription_id=None,
+                        management_group_id=None, location=None, resource_group=None,
+                        deployment_name=None):
+    ensure_bicep_installation(cmd.cli_ctx, stdout=False)
+
+    minimum_supported_version = "0.41.2"
+    if bicep_version_greater_than_or_equal_to(cmd.cli_ctx, minimum_supported_version):
+        args = ["snapshot", file]
+        if mode:
+            args += ["--mode", mode]
+        if tenant_id:
+            args += ["--tenant-id", tenant_id]
+        if subscription_id:
+            args += ["--subscription-id", subscription_id]
+        if management_group_id:
+            args += ["--management-group-id", management_group_id]
+        if location:
+            args += ["--location", location]
+        if resource_group:
+            args += ["--resource-group", resource_group]
+        if deployment_name:
+            args += ["--deployment-name", deployment_name]
+
+        output = run_bicep_command(cmd.cli_ctx, args)
+
+        if output:
+            print(output)
+    else:
+        raise ValidationError(
+            f"az bicep snapshot could not be executed with the current version of Bicep CLI. "
+            f"Please upgrade Bicep CLI to v{minimum_supported_version} or later."
+        )
+
+
+def run_bicep_cli_passthrough(cmd, command_string):
+    import shlex
+
+    ensure_bicep_installation(cmd.cli_ctx, stdout=False)
+
+    # Use non-POSIX mode so that backslashes in Windows paths are preserved.
+    # In non-POSIX mode, shlex retains the surrounding quotes on quoted tokens,
+    # so strip them so the values are passed through cleanly to the Bicep CLI.
+    args = []
+    for token in shlex.split(command_string, posix=False):
+        if len(token) >= 2 and token[0] in ('"', "'") and token[0] == token[-1]:
+            token = token[1:-1]
+        args.append(token)
+
+    if not args:
+        raise InvalidArgumentValueError("--command must not be empty.")
+
+    output = run_bicep_command(cmd.cli_ctx, args)
+
+    if output:
+        print(output)
+
+
 def create_resourcemanager_privatelink(
         cmd, resource_group, name, location):
     rcf = _resource_privatelinks_client_factory(cmd.cli_ctx)

src/azure-cli/azure/cli/command_modules/resource/tests/latest/test_resource.py

@@ -5669,6 +5669,68 @@ def test_bicep_lint_diagnostics_format_sarif(self):
 
         self.cmd('az bicep lint -f {tf} --diagnostics-format sarif')
 
+
+class BicepSnapshotTest(LiveScenarioTest):
+    def setUp(self):
+        super().setUp()
+        self.cmd('az bicep uninstall')
+
+    def tearDown(self):
+        super().tearDown()
+        self.cmd('az bicep uninstall')
+
+    def test_bicep_snapshot(self):
+        curr_dir = os.path.dirname(os.path.realpath(__file__))
+        params_file = os.path.join(curr_dir, 'sample_params.bicepparam').replace('\\', '\\\\')
+        snapshot_path = os.path.join(curr_dir, 'sample_params.snapshot.json')
+        self.kwargs.update({
+            'pf': params_file,
+        })
+
+        try:
+            # Capture (default mode).
+            self.cmd('az bicep snapshot --file {pf}')
+            self.assertTrue(os.path.exists(snapshot_path))
+
+            # Validate against the just-captured snapshot.
+            self.cmd('az bicep snapshot --file {pf} --mode Validate')
+        finally:
+            if os.path.exists(snapshot_path):
+                os.remove(snapshot_path)
+
+
+class BicepRunTest(LiveScenarioTest):
+    def setUp(self):
+        super().setUp()
+        self.cmd('az bicep uninstall')
+
+    def tearDown(self):
+        super().tearDown()
+        self.cmd('az bicep uninstall')
+
+    def test_bicep_run_version(self):
+        # Ensure Bicep CLI is installed so the passthrough has something to call.
+        self.cmd('az bicep install')
+        # Use the --option=value form because the value itself starts with --,
+        # which argparse otherwise treats as another option flag.
+        self.cmd('az bicep run --command=--version')
+
+    def test_bicep_run_build(self):
+        curr_dir = os.path.dirname(os.path.realpath(__file__))
+        bf = os.path.join(curr_dir, 'sample_params.bicep').replace('\\', '\\\\')
+        self.kwargs.update({
+            'bf': bf,
+        })
+
+        self.cmd('az bicep install')
+        self.cmd('az bicep run --command "build {bf} --stdout"')
+
+    def test_bicep_run_empty_command_fails(self):
+        from azure.cli.core.azclierror import InvalidArgumentValueError
+        with self.assertRaises(InvalidArgumentValueError):
+            self.cmd('az bicep run --command "   "')
+
+
 class BicepInstallationTest(LiveScenarioTest):
     def setup(self):
         super().setup()

src/azure-cli/azure/cli/command_modules/resource/tests/latest/test_resource_bicep.py

@@ -19,7 +19,11 @@
     _get_bicep_download_url,
     _bicep_version_check_file_path,
 )
-from azure.cli.core.azclierror import InvalidTemplateError
+from azure.cli.command_modules.resource.custom import (
+    run_bicep_cli_passthrough,
+    snapshot_bicep_file,
+)
+from azure.cli.core.azclierror import InvalidArgumentValueError, InvalidTemplateError
 from azure.cli.core.mock import DummyCli
 
 
@@ -302,4 +306,128 @@ def test_bicep_version_greater_than_or_equal_to_use_cli_managed_binary(self, use
         result = bicep_version_greater_than_or_equal_to(self.cli_ctx, "0.13.2")
 
         self.assertFalse(result)
-        run_command_mock.assert_called_once_with(".azure/bin/bicep", ["--version"])
+        run_command_mock.assert_called_once_with(".azure/bin/bicep", ["--version"])
+
+
+class TestBicepSnapshot(unittest.TestCase):
+    def setUp(self):
+        self.cli_ctx = DummyCli(random_config_dir=True)
+        self.cmd = mock.Mock()
+        self.cmd.cli_ctx = self.cli_ctx
+
+    @mock.patch("azure.cli.command_modules.resource.custom.run_bicep_command")
+    @mock.patch("azure.cli.command_modules.resource.custom.bicep_version_greater_than_or_equal_to")
+    @mock.patch("azure.cli.command_modules.resource.custom.ensure_bicep_installation")
+    def test_snapshot_bicep_file_passes_minimum_args(
+        self, ensure_bicep_installation_mock, bicep_version_check_mock, run_bicep_command_mock
+    ):
+        bicep_version_check_mock.return_value = True
+        run_bicep_command_mock.return_value = ""
+
+        snapshot_bicep_file(self.cmd, "main.bicepparam")
+
+        ensure_bicep_installation_mock.assert_called_once_with(self.cli_ctx, stdout=False)
+        bicep_version_check_mock.assert_called_once_with(self.cli_ctx, "0.41.2")
+        run_bicep_command_mock.assert_called_once_with(self.cli_ctx, ["snapshot", "main.bicepparam"])
+
+    @mock.patch("azure.cli.command_modules.resource.custom.run_bicep_command")
+    @mock.patch("azure.cli.command_modules.resource.custom.bicep_version_greater_than_or_equal_to")
+    @mock.patch("azure.cli.command_modules.resource.custom.ensure_bicep_installation")
+    def test_snapshot_bicep_file_passes_all_optional_args(
+        self, ensure_bicep_installation_mock, bicep_version_check_mock, run_bicep_command_mock
+    ):
+        bicep_version_check_mock.return_value = True
+        run_bicep_command_mock.return_value = ""
+
+        snapshot_bicep_file(
+            self.cmd,
+            "main.bicepparam",
+            mode="Validate",
+            tenant_id="tenant-id",
+            subscription_id="sub-id",
+            management_group_id="mg-id",
+            location="westus",
+            resource_group="myRg",
+            deployment_name="myDeployment",
+        )
+
+        run_bicep_command_mock.assert_called_once_with(
+            self.cli_ctx,
+            [
+                "snapshot",
+                "main.bicepparam",
+                "--mode", "Validate",
+                "--tenant-id", "tenant-id",
+                "--subscription-id", "sub-id",
+                "--management-group-id", "mg-id",
+                "--location", "westus",
+                "--resource-group", "myRg",
+                "--deployment-name", "myDeployment",
+            ],
+        )
+
+    @mock.patch("azure.cli.command_modules.resource.custom.run_bicep_command")
+    @mock.patch("azure.cli.command_modules.resource.custom.bicep_version_greater_than_or_equal_to")
+    @mock.patch("azure.cli.command_modules.resource.custom.ensure_bicep_installation")
+    def test_snapshot_bicep_file_errors_when_bicep_too_old(
+        self, ensure_bicep_installation_mock, bicep_version_check_mock, run_bicep_command_mock
+    ):
+        from azure.cli.core.azclierror import ValidationError
+
+        bicep_version_check_mock.return_value = False
+
+        with self.assertRaisesRegex(ValidationError, "az bicep snapshot.*0\\.41\\.2"):
+            snapshot_bicep_file(self.cmd, "main.bicepparam")
+
+        run_bicep_command_mock.assert_not_called()
+
+
+class TestBicepRun(unittest.TestCase):
+    def setUp(self):
+        self.cli_ctx = DummyCli(random_config_dir=True)
+        self.cmd = mock.Mock()
+        self.cmd.cli_ctx = self.cli_ctx
+
+    @mock.patch("azure.cli.command_modules.resource.custom.run_bicep_command")
+    @mock.patch("azure.cli.command_modules.resource.custom.ensure_bicep_installation")
+    def test_run_bicep_cli_passthrough_forwards_split_args(self, ensure_bicep_installation_mock, run_bicep_command_mock):
+        run_bicep_command_mock.return_value = ""
+
+        run_bicep_cli_passthrough(self.cmd, "build main.bicep --stdout")
+
+        ensure_bicep_installation_mock.assert_called_once_with(self.cli_ctx, stdout=False)
+        run_bicep_command_mock.assert_called_once_with(
+            self.cli_ctx, ["build", "main.bicep", "--stdout"]
+        )
+
+    @mock.patch("azure.cli.command_modules.resource.custom.run_bicep_command")
+    @mock.patch("azure.cli.command_modules.resource.custom.ensure_bicep_installation")
+    def test_run_bicep_cli_passthrough_preserves_quoted_args(self, ensure_bicep_installation_mock, run_bicep_command_mock):
+        run_bicep_command_mock.return_value = ""
+
+        run_bicep_cli_passthrough(self.cmd, 'build "path with spaces/main.bicep"')
+
+        run_bicep_command_mock.assert_called_once_with(
+            self.cli_ctx, ["build", "path with spaces/main.bicep"]
+        )
+
+    @mock.patch("azure.cli.command_modules.resource.custom.run_bicep_command")
+    @mock.patch("azure.cli.command_modules.resource.custom.ensure_bicep_installation")
+    def test_run_bicep_cli_passthrough_preserves_windows_path_backslashes(self, ensure_bicep_installation_mock, run_bicep_command_mock):
+        run_bicep_command_mock.return_value = ""
+
+        # Windows paths use backslashes which collide with POSIX shell escape semantics.
+        # The passthrough must preserve them so the Bicep CLI receives a valid path.
+        run_bicep_cli_passthrough(self.cmd, r"build D:\azure-cli\samples\main.bicep --stdout")
+
+        run_bicep_command_mock.assert_called_once_with(
+            self.cli_ctx, ["build", r"D:\azure-cli\samples\main.bicep", "--stdout"]
+        )
+
+    @mock.patch("azure.cli.command_modules.resource.custom.run_bicep_command")
+    @mock.patch("azure.cli.command_modules.resource.custom.ensure_bicep_installation")
+    def test_run_bicep_cli_passthrough_raises_when_command_empty(self, ensure_bicep_installation_mock, run_bicep_command_mock):
+        with self.assertRaisesRegex(InvalidArgumentValueError, "--command must not be empty."):
+            run_bicep_cli_passthrough(self.cmd, "   ")
+
+        run_bicep_command_mock.assert_not_called()