Enable a deployed object list on the CLI

src/k8s-configuration/azext_k8s_configuration/_help.py

@@ -174,3 +174,24 @@
           --cluster-name mycluster --cluster-type connectedClusters --name myconfig \\
           --kustomization-name my-kustomization
 """
+
+helps['k8s-configuration flux deployed-object list'] = """
+    type: command
+    short-summary: List deployed objects associated with a Kubernetes Flux v2 Configuration.
+    examples:
+      - name: List all deployed objects associated with a Kubernetes Flux v2 Configuration on a cluster
+        text: |-
+          az k8s-configuration flux deployed-object list --resource-group my-resource-group \\
+          --cluster-name mycluster --name myconfig --cluster-type connectedClusters
+"""
+
+helps['k8s-configuration flux deployed-object show'] = """
+    type: command
+    short-summary: Show a deployed object associated with a Flux v2 Configuration.
+    examples:
+      - name: Show details of a deployed object associated with a Kubernetes Flux v2 Configuration
+        text: |-
+          az k8s-configuration flux deployed-object show --resource-group my-resource-group \\
+          --cluster-name mycluster --cluster-type connectedClusters --name myconfig \\
+          --object-name my-object --object-namespace my-namespace --object-kind GitRepository
+"""

src/k8s-configuration/azext_k8s_configuration/_params.py

@@ -118,6 +118,15 @@ def load_arguments(self, _):
                    options_list=['--yes', '-y'],
                    help='Do not prompt for confirmation')
 
+    with self.argument_context('k8s-configuration flux deployed-object show') as c:
+        c.argument('object_name',
+                   help='Name of the object deployed by the configuration on the cluster.')
+        c.argument('object_namespace',
+                   help='Namespace of the object deployed by the configuration on the cluster.')
+        c.argument('object_kind',
+                   arg_type=get_enum_type(['GitRepository', 'Bucket', 'HelmRepository', 'HelmChart', 'HelmRelease', 'Kustomization']),
+                   help='Kind of the object deployed by the configuration on the cluster.')
+
     with self.argument_context('k8s-configuration') as c:
         c.argument('name',
                    options_list=['--name', '-n'],

src/k8s-configuration/azext_k8s_configuration/commands.py

@@ -10,6 +10,8 @@
     k8s_configuration_sourcecontrol_client
 )
 from .format import (
+    fluxconfig_deployed_object_list_table_format,
+    fluxconfig_deployed_object_show_table_format,
     fluxconfig_list_table_format,
     fluxconfig_show_table_format,
     fluxconfig_kustomization_list_table_format,
@@ -44,6 +46,10 @@ def load_command_table(self, _):
         g.custom_command('list', 'flux_config_list_kustomization', table_transformer=fluxconfig_kustomization_list_table_format)
         g.custom_show_command('show', 'flux_config_show_kustomization', table_transformer=fluxconfig_kustomization_show_table_format)
 
+    with self.command_group('k8s-configuration flux deployed-object', k8s_configuration_fluxconfig_sdk, client_factory=k8s_configuration_fluxconfig_client, is_preview=True) as g:
+        g.custom_command('list', 'flux_config_list_deployed_object', table_transformer=fluxconfig_deployed_object_list_table_format)
+        g.custom_show_command('show', 'flux_config_show_deployed_object', table_transformer=fluxconfig_deployed_object_show_table_format)
+
     with self.command_group('k8s-configuration', k8s_configuration_sourcecontrol_sdk, client_factory=k8s_configuration_sourcecontrol_client) as g:
         g.custom_command('create', 'sourcecontrol_create', deprecate_info=self.deprecate(redirect='k8s-configuration flux create'))
         g.custom_command('list', 'sourcecontrol_list', table_transformer=sourcecontrol_list_table_format, deprecate_info=self.deprecate(redirect='k8s-configuration flux list'))

src/k8s-configuration/azext_k8s_configuration/consts.py

@@ -64,6 +64,9 @@
 SHOW_KUSTOMIZATION_NO_EXIST_ERROR = "Error! Kustomization with name '{0}' does not exist on configuration '{1}'."
 SHOW_KUSTOMIZATION_NO_EXIST_HELP = "You can view all kustomizations on a configuration with 'az k8s-configuration flux kustomization list'"
 
+SHOW_DEPLOYED_OBJECT_NO_EXIST_ERROR = "Error! Deployed object with name '{0}', namespace '{1}', and kind '{2}' does not exist on configuration '{3}'."
+SHOW_DEPLOYED_OBJECT_NO_EXIST_HELP = "You can view all deployed objects on a configuration with 'az k8s-configuration flux deployed-object list'"
+
 SSH_PRIVATE_KEY_WITH_HTTP_URL_ERROR = "Error! An --ssh-private-key cannot be used with an http(s) url"
 SSH_PRIVATE_KEY_WITH_HTTP_URL_HELP = "Verify the url provided is a valid ssh url and not an http(s) url"
 

src/k8s-configuration/azext_k8s_configuration/custom.py

@@ -121,6 +121,20 @@ def flux_config_show_kustomization(cmd, client, resource_group_name, cluster_typ
     return provider.show_kustomization(resource_group_name, cluster_type, cluster_name, name, kustomization_name)
 
 
+def flux_config_list_deployed_object(cmd, client, resource_group_name, cluster_type, cluster_name, name):
+
+    provider = FluxConfigurationProvider(cmd)
+    return provider.list_deployed_object(resource_group_name, cluster_type, cluster_name, name)
+
+
+def flux_config_show_deployed_object(cmd, client, resource_group_name, cluster_type, cluster_name, name,
+                                     object_name, object_namespace, object_kind):
+
+    provider = FluxConfigurationProvider(cmd)
+    return provider.show_deployed_object(resource_group_name, cluster_type, cluster_name, name,
+                                         object_name, object_namespace, object_kind)
+
+
 def flux_config_delete(cmd, client, resource_group_name, cluster_type,
                        cluster_name, name, force=False, no_wait=False, yes=False):
     provider = FluxConfigurationProvider(cmd)

src/k8s-configuration/azext_k8s_configuration/format.py

@@ -68,3 +68,24 @@ def __get_fluxconfig_kustomization_table_row(key, value):
         ('prune', value['prune']),
         ('force', value['force'])
     ])
+
+
+def fluxconfig_deployed_object_list_table_format(results):
+    return [__get_fluxconfig_deployed_object_table_row(result) for result in results]
+
+
+def fluxconfig_deployed_object_show_table_format(result):
+    return __get_fluxconfig_deployed_object_table_row(result)
+
+
+def __get_fluxconfig_deployed_object_table_row(result):
+    applied_by = 'None'
+    if result['appliedBy']:
+        applied_by = result['appliedBy']['namespace'] + result['appliedBy']['name']
+    return OrderedDict([
+        ('kind', result['kind']),
+        ('name', result['name']),
+        ('namespace', result['namespace']),
+        ('complianceState', result['complianceState']),
+        ('appliedBy', applied_by)
+    ])

src/k8s-configuration/azext_k8s_configuration/providers/FluxConfigurationProvider.py

@@ -336,6 +336,26 @@ def show_kustomization(self, resource_group_name, cluster_type, cluster_name, na
             )
         return {kustomization_name: current_config.kustomizations[kustomization_name]}
 
+    def list_deployed_object(self, resource_group_name, cluster_type, cluster_name, name):
+        # Determine ClusterRP
+        cluster_rp = get_cluster_rp(cluster_type)
+        current_config = self.client.get(resource_group_name, cluster_rp, cluster_type, cluster_name, name)
+        return current_config.statuses
+
+    def show_deployed_object(self, resource_group_name, cluster_type, cluster_name, name,
+                             object_name, object_namespace, object_kind):
+        # Determine ClusterRP
+        cluster_rp = get_cluster_rp(cluster_type)
+        current_config = self.client.get(resource_group_name, cluster_rp, cluster_type, cluster_name, name)
+
+        for status in current_config.statuses:
+            if status.name == object_name and status.namespace == object_namespace and status.kind == object_kind:
+                return status
+        raise ValidationError(
+            consts.SHOW_DEPLOYED_OBJECT_NO_EXIST_ERROR.format(object_name, object_namespace, object_kind, name),
+            consts.SHOW_DEPLOYED_OBJECT_NO_EXIST_HELP
+        )
+
     def delete(self, resource_group_name, cluster_type, cluster_name, name, force, no_wait, yes):
         # Confirmation message for deletes
         user_confirmation_factory(self.cmd, yes)

testing/pipeline/templates/run-test.yml

@@ -109,4 +109,4 @@ jobs:
       inlineScript: |
         .\Cleanup.ps1 -CI
       workingDirectory: $(TEST_PATH)
-    condition: succeededOrFailed()
+    condition: always()