edgeoperator: az aldo system-readiness show (preview)

.github/CODEOWNERS

@@ -345,3 +345,4 @@
 /src/aimanager/ @ximengzhao
 
 /src/fileshare/ @ankushb
+/src/edgeoperator/ @kleyvaortega

src/edgeoperator/HISTORY.rst

@@ -0,0 +1,6 @@
+Release History
+===============
+
+0.1.0
+++++++
+* Initial preview release with the system readiness show command.

src/edgeoperator/README.rst

@@ -0,0 +1,8 @@
+Microsoft Azure CLI ``edgeoperator`` Extension
+==============================================
+
+This package provides Azure CLI commands for Azure Local Disconnected Operations.
+
+Use ``az aldo system-readiness show`` to retrieve the current system readiness status.
+
+The command uses the public `Microsoft.EdgeOperator SystemReadiness REST API specification <https://github.com/Azure/azure-rest-api-specs/tree/main/specification/edgeoperator/resource-manager/Microsoft.EdgeOperator/SystemReadiness>`_.

src/edgeoperator/azext_edgeoperator/__init__.py

@@ -0,0 +1,31 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from azure.cli.core import AzCommandsLoader
+
+from azext_edgeoperator._help import helps  # pylint: disable=unused-import
+
+
+class AldoCommandsLoader(AzCommandsLoader):
+
+    def __init__(self, cli_ctx=None):
+        from azure.cli.core.commands import CliCommandType
+
+        command_type = CliCommandType(operations_tmpl="azext_edgeoperator.custom#{}")
+        super().__init__(cli_ctx=cli_ctx, custom_command_type=command_type)
+
+    def load_command_table(self, args):
+        from azext_edgeoperator.commands import load_command_table
+
+        load_command_table(self, args)
+        return self.command_table
+
+    def load_arguments(self, command):
+        from azext_edgeoperator._params import load_arguments
+
+        load_arguments(self, command)
+
+
+COMMAND_LOADER_CLS = AldoCommandsLoader

src/edgeoperator/azext_edgeoperator/_help.py

@@ -0,0 +1,23 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from knack.help_files import helps
+
+
+helps["aldo"] = """
+    type: group
+    short-summary: Manage Azure Local Disconnected Operations resources.
+"""
+
+helps["aldo system-readiness"] = """
+    type: group
+    short-summary: Manage ALDO system readiness.
+"""
+
+helps["aldo system-readiness show"] = """
+    type: command
+    short-summary: Show the current ALDO system readiness status.
+    long-summary: Retrieves the read-only system readiness singleton for the current subscription.
+"""

src/edgeoperator/azext_edgeoperator/_params.py

@@ -0,0 +1,10 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+# pylint: disable=unused-argument
+
+
+def load_arguments(self, _):
+    pass

src/edgeoperator/azext_edgeoperator/azext_metadata.json

@@ -0,0 +1,4 @@
+{
+    "azext.isPreview": true,
+    "azext.minCliCoreVersion": "2.75.0"
+}

src/edgeoperator/azext_edgeoperator/commands.py

@@ -0,0 +1,8 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+def load_command_table(self, _):
+    with self.command_group("aldo system-readiness") as group:
+        group.custom_show_command("show", "show_system_readiness")

src/edgeoperator/azext_edgeoperator/custom.py

@@ -0,0 +1,24 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from azure.cli.core.commands.client_factory import get_subscription_id
+from azure.cli.core.util import send_raw_request
+
+
+API_VERSION = "2026-06-01-preview"
+
+
+def _system_readiness_url(cmd):
+    endpoint = cmd.cli_ctx.cloud.endpoints.resource_manager.rstrip("/")
+    subscription_id = get_subscription_id(cmd.cli_ctx)
+    return (
+        "{}/subscriptions/{}/providers/Microsoft.EdgeOperator/"
+        "systemReadiness/default?api-version={}"
+    ).format(endpoint, subscription_id, API_VERSION)
+
+
+def show_system_readiness(cmd):
+    response = send_raw_request(cmd.cli_ctx, "GET", _system_readiness_url(cmd))
+    return response.json()

src/edgeoperator/azext_edgeoperator/tests/latest/recordings/test_system_readiness_show.yaml

@@ -0,0 +1,29 @@
+interactions:
+- request:
+    body: null
+    headers:
+      Accept:
+      - '*/*'
+      Accept-Encoding:
+      - gzip, deflate
+      CommandName:
+      - aldo system-readiness show
+      Connection:
+      - keep-alive
+      User-Agent:
+      - AZURECLI
+    method: GET
+    uri: https://management.azure.com/subscriptions/00000000-0000-0000-0000-000000000000/providers/Microsoft.EdgeOperator/systemReadiness/default?api-version=2026-06-01-preview
+  response:
+    body:
+      string: '{"id":"/subscriptions/00000000-0000-0000-0000-000000000000/providers/Microsoft.EdgeOperator/systemReadiness/default","name":"default","type":"Microsoft.EdgeOperator/systemReadiness","properties":{"systemReady":false,"categories":[{"categoryName":"services","readinessPercentage":100,"errorMessageDetails":[]},{"categoryName":"diagnostics","readinessPercentage":0,"errorMessageDetails":[]},{"categoryName":"identity","readinessPercentage":100,"errorMessageDetails":[]},{"categoryName":"networking","readinessPercentage":50,"errorMessageDetails":[]}]}}'
+    headers:
+      content-length:
+      - '550'
+      content-type:
+      - application/json; charset=utf-8
+    status:
+      code: 200
+      message: OK
+version: 1
+

src/edgeoperator/azext_edgeoperator/tests/latest/test_custom.py

@@ -0,0 +1,34 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from unittest import TestCase
+from unittest.mock import Mock, patch
+
+from azext_edgeoperator.custom import API_VERSION, show_system_readiness
+
+
+class SystemReadinessCustomTest(TestCase):
+
+    @patch("azext_edgeoperator.custom.send_raw_request")
+    @patch("azext_edgeoperator.custom.get_subscription_id")
+    def test_show_uses_singleton_resource(self, get_subscription_id, send_raw_request):
+        get_subscription_id.return_value = "00000000-0000-0000-0000-000000000000"
+        response = Mock()
+        response.json.return_value = {"properties": {"systemReady": True}}
+        send_raw_request.return_value = response
+
+        cli_ctx = Mock()
+        cli_ctx.cloud.endpoints.resource_manager = "https://management.example.com/"
+        cmd = Mock(cli_ctx=cli_ctx)
+
+        result = show_system_readiness(cmd)
+
+        expected_url = (
+            "https://management.example.com/subscriptions/"
+            "00000000-0000-0000-0000-000000000000/providers/Microsoft.EdgeOperator/"
+            "systemReadiness/default?api-version={}"
+        ).format(API_VERSION)
+        send_raw_request.assert_called_once_with(cli_ctx, "GET", expected_url)
+        self.assertTrue(result["properties"]["systemReady"])

src/edgeoperator/azext_edgeoperator/tests/latest/test_system_readiness_scenario.py

@@ -0,0 +1,64 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import re
+
+from azure.cli.testsdk import JMESPathCheck, ScenarioTest
+from azure.cli.testsdk.scenario_tests import RecordingProcessor
+
+# Fixed host used in the recording. The command derives the ARM endpoint from the
+# active cloud, which differs between environments (ALDO locally, AzureCloud in CI),
+# so both recording and playback requests are normalized to this host.
+NORMALIZED_HOST = "https://management.azure.com"
+_READINESS_PATH = re.compile(
+    r"^https?://[^/]+(/subscriptions/.+?/providers/Microsoft\.EdgeOperator/systemReadiness/.*)$"
+)
+
+
+class PrivateAuthRequestFilter(RecordingProcessor):
+    """Drop ALDO private-cloud token requests/responses when recording live."""
+
+    def process_request(self, request):
+        if "autonomous.cloud.private" in request.uri and "systemReadiness" not in request.uri:
+            return None
+        return request
+
+
+class ArmEndpointNormalizer(RecordingProcessor):
+    """Normalize the ARM host on the systemReadiness request to a fixed value."""
+
+    def process_request(self, request):
+        request.uri = _READINESS_PATH.sub(NORMALIZED_HOST + r"\1", request.uri)
+        return request
+
+
+class SystemReadinessScenarioTest(ScenarioTest):
+
+    def __init__(self, method_name, **kwargs):
+        recording_processors = kwargs.pop("recording_processors", [])
+        replay_processors = kwargs.pop("replay_processors", [])
+        super().__init__(
+            method_name,
+            recording_processors=[
+                PrivateAuthRequestFilter(),
+                ArmEndpointNormalizer(),
+            ] + recording_processors,
+            replay_processors=[ArmEndpointNormalizer()] + replay_processors,
+            **kwargs
+        )
+
+    def test_system_readiness_show(self):
+        result = self.cmd("aldo system-readiness show").assert_with_checks([
+            JMESPathCheck("name", "default"),
+            JMESPathCheck("type", "Microsoft.EdgeOperator/systemReadiness"),
+        ]).get_output_in_json()
+
+        self.assertIsInstance(result["properties"]["systemReady"], bool)
+        categories = result["properties"]["categories"]
+        self.assertIsInstance(categories, list)
+        for category in categories:
+            self.assertIn("categoryName", category)
+            self.assertIn("readinessPercentage", category)
+            self.assertIsInstance(category["errorMessageDetails"], list)

src/edgeoperator/setup.cfg

@@ -0,0 +1 @@
+#setup.cfg

src/edgeoperator/setup.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python
+
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from codecs import open
+from setuptools import find_packages, setup
+
+VERSION = "0.1.0"
+
+with open("README.rst", "r", encoding="utf-8") as f:
+    README = f.read()
+with open("HISTORY.rst", "r", encoding="utf-8") as f:
+    HISTORY = f.read()
+
+setup(
+    name="edgeoperator",
+    version=VERSION,
+    description="Microsoft Azure Command-Line Tools ALDO Extension",
+    author="Microsoft Corporation",
+    author_email="azpycli@microsoft.com",
+    url="https://github.com/Azure/azure-cli-extensions/tree/main/src/edgeoperator",
+    long_description=README + "\n\n" + HISTORY,
+    license="MIT",
+    packages=find_packages(exclude=["*.tests", "*.tests.*"]),
+    install_requires=[],
+    package_data={"azext_edgeoperator": ["azext_metadata.json"]},
+)

src/service_name.json

@@ -1093,5 +1093,15 @@
     "Command": "az interconnect-block",
     "AzureServiceName": "Interconnect",
     "URL": ""
+  },
+  {
+    "Command": "az aldo",
+    "AzureServiceName": "Azure Local Disconnected Operations",
+    "URL": ""
+  },
+  {
+    "Command": "az aldo system-readiness",
+    "AzureServiceName": "Azure Local Disconnected Operations",
+    "URL": ""
   }
-]
+]