[load] Adding commands to manage notification-rule in azure load testing using CLI (#8622)

* Adding commands for scheduling

* added list schedule command

* initial tests for scheduling

* Updated tests

* Enhancing test cases for schedule

* Optimizing utils and addressing PR comments

* Adding invalid test cases

* Addressing PR comments

* linting and styling

* test cases and recordings updated

* nit

* Updating version

* nit

* Addressing PR comments and issues after bug bash

* Fixes after running githooks

* removed datetime.UTC as not supported in python 3.9

* Working code for notification command

* nit

* Changes along with test cases for notification rule

* Updating version to 2.0.0

* Changes after bug bash

* linter fix

Author: Himanshu49

src/load/HISTORY.rst

@@ -2,10 +2,17 @@
 
 Release History
 ===============
+
+2.0.0
+++++++
+* Add commands for creating and managing notification rules using CLI. 
+
+
 1.8.0
 ++++++
 * Add commands for creating and managing schedule triggers using CLI. 
 
+
 1.7.0
 ++++++
 * Add support for metrics reference identity. Metrics reference identity can be set using `--metrics-reference-identity` argument in 'az load test create' and 'az load test update' commands. Metrics reference identity set in YAML config file under key `referenceIdentities` with `kind` as `Metrics` will also be honoured.

src/load/azext_load/__init__.py

@@ -19,6 +19,7 @@ def load_command_table(self, args):
         from azext_load.data_plane.load_test.commands import load_test_commands
         from azext_load.data_plane.load_test_run.commands import load_test_run_commands
         from azext_load.data_plane.load_trigger.commands import load_trigger_schedule_commands
+        from azext_load.data_plane.load_notifications.commands import load_notification_commands
         from azure.cli.core.aaz import load_aaz_command_table
         try:
             from . import aaz
@@ -34,6 +35,7 @@ def load_command_table(self, args):
         load_test_commands(self, args)
         load_test_run_commands(self, args)
         load_trigger_schedule_commands(self, args)
+        load_notification_commands(self, args)
         return self.command_table
 
     def load_arguments(self, command):
@@ -42,12 +44,14 @@ def load_arguments(self, command):
         from azext_load.data_plane.load_test.params import load_arguments as load_test_arguments
         from azext_load.data_plane.load_test_run.params import load_arguments as load_test_run_arguments
         from azext_load.data_plane.load_trigger.params import load_arguments as load_trigger_schedule_arguments
+        from azext_load.data_plane.load_notifications.params import load_arguments as load_notification_arguments
 
         load_arguments(self, command)
         load_common_arguments(self, command)
         load_test_arguments(self, command)
         load_test_run_arguments(self, command)
         load_trigger_schedule_arguments(self, command)
+        load_notification_arguments(self, command)
 
 
 COMMAND_LOADER_CLS = LoadCommandsLoader

src/load/azext_load/_help.py

@@ -13,8 +13,10 @@
 from azext_load.data_plane.load_test.help import helps as data_load_test_helps
 from azext_load.data_plane.load_test_run.help import helps as data_load_test_run_helps
 from azext_load.data_plane.load_trigger.help import helps as data_load_trigger_helps
+from azext_load.data_plane.load_notifications.help import helps as data_load_notifications_helps
 
 helps.update(data_common_helps)
 helps.update(data_load_test_helps)
 helps.update(data_load_test_run_helps)
 helps.update(data_load_trigger_helps)
+helps.update(data_load_notifications_helps)

src/load/azext_load/data_plane/help.py

@@ -100,3 +100,11 @@
 short-summary: Command group to manage schedule triggers.
 long-summary: Command group to manage schedule triggers.
 """ + _common_params
+
+helps[
+    "load notification-rule"
+] = """
+type: group
+short-summary: Command group to manage notification rules.
+long-summary: Command group to manage notification rules.
+""" + _common_params

src/load/azext_load/data_plane/load_notifications/__init__.py

@@ -0,0 +1,6 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+__path__ = __import__("pkgutil").extend_path(__path__, __name__)

src/load/azext_load/data_plane/load_notifications/commands.py

@@ -0,0 +1,22 @@
+# --------------------------------------------------------------------------------------------
+# 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 import CliCommandType
+
+admin_custom_sdk = CliCommandType(
+    operations_tmpl="azext_load.data_plane.load_notifications.custom#{}"
+)
+
+
+def load_notification_commands(self, _):
+
+    with self.command_group(
+        "load notification-rule", custom_command_type=admin_custom_sdk, is_preview=True
+    ) as g:
+        g.custom_command("create", "create_notification_rule")
+        g.custom_command("update", "update_notification_rule")
+        g.custom_show_command("show", "show_notification_rule")
+        g.custom_command("list", "list_notification_rules")
+        g.custom_command("delete", "delete_notification_rule", confirmation=True)

src/load/azext_load/data_plane/load_notifications/custom.py

@@ -0,0 +1,128 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from knack.log import get_logger
+from azext_load.data_plane.utils.utils import (
+    get_admin_data_plane_client)
+from azext_load.data_plane.load_notifications import utils
+from azure.cli.core.azclierror import InvalidArgumentValueError
+from azure.core.exceptions import ResourceNotFoundError
+
+logger = get_logger(__name__)
+
+
+def create_notification_rule(
+    cmd,
+    load_test_resource,
+    action_groups,
+    notification_rule_id,
+    resource_group_name=None,
+    event=None,
+    display_name=None,
+    test_ids=None,
+    all_tests=False,
+    all_events=False,
+):
+    client = get_admin_data_plane_client(cmd, load_test_resource, resource_group_name)
+    body = None
+    try:
+        body = client.get_notification_rule(notification_rule_id)
+    except ResourceNotFoundError:
+        pass
+
+    if body is not None:
+        msg = "Notification rule with given ID: %s already exists." % notification_rule_id
+        logger.debug(msg)
+        raise InvalidArgumentValueError(msg)
+
+    logger.info("Creating notification rule.")
+    notification_rule = utils.get_notification_rule_create_body(
+        action_groups, event, display_name, test_ids, all_tests, all_events
+    )
+    logger.info("Notification rule object to be sent for create: %s", notification_rule)
+    response = client.create_or_update_notification_rule(notification_rule_id, notification_rule)
+    logger.info("Notification rule created successfully.")
+    return response.as_dict()
+
+
+def update_notification_rule(
+    cmd,
+    load_test_resource,
+    notification_rule_id,
+    resource_group_name=None,
+    action_groups=None,
+    add_event=None,
+    remove_event=None,
+    display_name=None,
+    test_ids=None,
+    all_tests=False,
+):
+    client = get_admin_data_plane_client(cmd, load_test_resource, resource_group_name)
+    logger.info("Updating notification rule.")
+    try:
+        existing_notification_rule = client.get_notification_rule(notification_rule_id)
+    except ResourceNotFoundError:
+        msg = "Notification rule with given ID: %s does not exist." % notification_rule_id
+        logger.debug(msg)
+        raise InvalidArgumentValueError(msg)
+    logger.info("Existing notification rule: %s", existing_notification_rule)
+    new_notification_rule = utils.get_notification_rule_update_body(
+        existing_notification_rule,
+        action_groups,
+        add_event,
+        remove_event,
+        display_name,
+        test_ids,
+        all_tests,
+    )
+    logger.info("Incoming changes in notification rule: %s", new_notification_rule)
+    response = client.create_or_update_notification_rule(notification_rule_id, new_notification_rule)
+    logger.info("Notification rule updated successfully.")
+    return response.as_dict()
+
+
+def show_notification_rule(
+    cmd,
+    load_test_resource,
+    notification_rule_id,
+    resource_group_name=None,
+):
+    client = get_admin_data_plane_client(cmd, load_test_resource, resource_group_name)
+    logger.info(
+        "Getting notification rule with id: %s", notification_rule_id
+    )
+    response = client.get_notification_rule(notification_rule_id)
+    logger.debug("Fetched notification rule: %s", response)
+    return response.as_dict()
+
+
+def list_notification_rules(
+    cmd,
+    load_test_resource,
+    resource_group_name=None,
+    test_ids=None,
+):
+    logger.info("Listing notification rules.")
+    client = get_admin_data_plane_client(cmd, load_test_resource, resource_group_name)
+    if test_ids:
+        test_ids = ",".join(test_ids)
+        logger.info("Filtering notification rules by test ids: %s", test_ids)
+    responses = client.list_notification_rule(test_ids=test_ids)
+    logger.info("Retrieved notification rules: %s", responses)
+    return [response.as_dict() for response in responses]
+
+
+def delete_notification_rule(
+    cmd,
+    load_test_resource,
+    notification_rule_id,
+    resource_group_name=None,
+):
+    logger.info(
+        "Deleting notification rule with id: %s", notification_rule_id
+    )
+    client = get_admin_data_plane_client(cmd, load_test_resource, resource_group_name)
+    client.delete_notification_rule(notification_rule_id)
+    logger.info("Deleted notification rule.")

src/load/azext_load/data_plane/load_notifications/help.py

@@ -0,0 +1,76 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+# pylint: disable=line-too-long
+# pylint: disable=too-many-lines
+
+helps = {}
+
+helps[
+    "load notification-rule create"
+] = """
+type: command
+short-summary: Create a new notification rule for load test resource.
+examples:
+    - name: Create a notification rule for all events.
+      text: |
+        az load notification-rule create --load-test-resource sample-alt-resource --resource-group sample-rg --action-groups /subscriptions/000000-0000-0000-0000-000000000000/resourcegroups/sample-rg/providers/microsoft.insights/actiongroups/sample-ag --notification-rule-id notification-sample-id --all-events
+    - name: Create a notification rule for all tests and TestRunEnded event.
+      text: |
+        az load notification-rule create --load-test-resource sample-alt-resource --resource-group sample-rg --action-groups /subscriptions/000000-0000-0000-0000-000000000000/resourcegroups/sample-rg/providers/microsoft.insights/actiongroups/sample-ag --notification-rule-id notification-sample-id --all-tests --event event-id=event1 type=TestRunEnded status=DONE,FAILED result=PASSED
+"""
+
+helps[
+    "load notification-rule update"
+] = """
+type: command
+short-summary: Update an existing notification rule for load test resource.
+examples:
+    - name: Update a notification rule enabled for all tests.
+      text: |
+        az load notification-rule update --load-test-resource sample-alt-resource --resource-group sample-rg --notification-rule-id notification-sample-id --all-tests
+    - name: Add a TestRunStarted event to an existing notification rule.
+      text: |
+        az load notification-rule update --load-test-resource sample-alt-resource --resource-group sample-rg --notification-rule-id notification-sample-id --add-event event-id=event1 type=TestRunStarted
+    - name: Remove an event from an existing notification rule and update the action group list.
+      text: |
+        az load notification-rule update --load-test-resource sample-alt-resource --resource-group sample-rg --notification-rule-id notification-sample-id --remove-event event-id=event1 --action-groups /subscriptions/000000-0000-0000-0000-000000000000/resourcegroups/sample-rg/providers/microsoft.insights/actiongroups/sample-ag
+"""
+
+helps[
+    "load notification-rule show"
+] = """
+type: command
+short-summary: Get the specified notification rule for load test resource.
+examples:
+    - name: Get a notification rule.
+      text: |
+        az load notification-rule show --load-test-resource sample-alt-resource --resource-group sample-rg --notification-rule-id notification-sample-id
+"""
+
+helps[
+    "load notification-rule delete"
+] = """
+type: command
+short-summary: Delete the specified notification rule for load test resource.
+examples:
+    - name: Delete a notification rule.
+      text: |
+        az load notification-rule delete --load-test-resource sample-alt-resource --resource-group sample-rg --notification-rule-id notification-sample-id --yes
+"""
+
+helps[
+    "load notification-rule list"
+] = """
+type: command
+short-summary: List all the notification rules for load test resource.
+examples:
+    - name: List all notification rules.
+      text: |
+        az load notification-rule list --load-test-resource sample-alt-resource --resource-group sample-rg
+    - name: List all notification rules for the specified test IDs.
+      text: |
+        az load notification-rule list --load-test-resource sample-alt-resource --resource-group sample-rg --test-ids sample-test-id
+"""

src/load/azext_load/data_plane/load_notifications/params.py

@@ -0,0 +1,49 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+# pylint: disable=too-many-lines
+# pylint: disable=too-many-statements
+# pylint: disable=line-too-long
+
+from azext_load.data_plane.utils import argtypes
+
+
+def load_arguments(self, _):
+    # Load Trigger Schedule
+    with self.argument_context("load notification-rule") as c:
+        c.argument("load_test_resource", argtypes.load_test_resource)
+        c.argument("resource_group_name", argtypes.resource_group)
+
+    # Load notification-rule create
+    with self.argument_context("load notification-rule create") as c:
+        c.argument("display_name", argtypes.notification_display_name)
+        c.argument("notification_rule_id", argtypes.notification_rule_id)
+        c.argument("test_ids", argtypes.notification_rule_test_ids)
+        c.argument("event", argtypes.notification_rule_event)
+        c.argument("action_groups", argtypes.action_groups)
+        c.argument("all_tests", argtypes.notification_all_tests)
+        c.argument("all_events", argtypes.notification_all_events)
+
+    # Load notification-rule update
+    with self.argument_context("load notification-rule update") as c:
+        c.argument("display_name", argtypes.notification_display_name)
+        c.argument("notification_rule_id", argtypes.notification_rule_id)
+        c.argument("test_ids", argtypes.notification_rule_test_ids)
+        c.argument("add_event", argtypes.notification_rule_add_event)
+        c.argument("remove_event", argtypes.notification_rule_remove_event)
+        c.argument("action_groups", argtypes.action_groups)
+        c.argument("all_tests", argtypes.notification_all_tests)
+
+    # Load notification-rule list
+    with self.argument_context("load notification-rule list") as c:
+        c.argument("test_ids", argtypes.notification_rule_test_ids)
+
+    # Load notification-rule show
+    with self.argument_context("load notification-rule show") as c:
+        c.argument("notification_rule_id", argtypes.notification_rule_id)
+
+    # Load notification-rule delete
+    with self.argument_context("load notification-rule delete") as c:
+        c.argument("notification_rule_id", argtypes.notification_rule_id)