improve Python unit testing and refactor _validators.py (#2120)

* refactor: apply Guard clauses in _validators.py to simplify code
* fix _validators.py trailing whitespace
* improve python test tructure, include execution of unit tests in make test-python
* crate make directive to run python unit tests, add unit tests for azext_aro._validators.validate_cidr
* add tests for test_validate_client_id and include test cases description
* add unit tests for validate_client_secret from azext_aro._validators
* add explicit fields to named tuple in test cases in test_validators.py
* add two test scenarios for validate_cluster_resource_group
* simplify mocks
* add test case for test_validate_cluster_resource_group
* improve test descriptions
* add test_validate_disk_encryption_set test to test validate_disk_encryption_set
* add test cases to test_validate_disk_encryption_set()
* refactor test_validator.py to use classes instead of namedtuples. Use mocks instead of specific defined classes
* refactor (simplify code): remove explicit assignemnt to None when it is the default value
* create test_validate_domain() with 1st test case
* add test case, domain with '_'
* explicit import of unittest.TestCase
* fix test message in test_validate_domain
* finish test_validate_domain()
* finish test_validate_sdn() and test_validate_pull_secret()
* create test_validate_subnet() with first test case
* finish test_validate_subnet() and minor refactor in _validators.py
* create test_validate_subnets() and add first test case
* finish validate_vnet_resource_group_name()
* finish test_validate_worker_count() of test_validators.py and simple refactor in _validators.py
* finish test_validate_worker_vm_disk_size_gb()
* refactor _validators.py
* add test_validate_refresh_cluster_credentials() and minor refactor of test_validators()
* refactor _test_validators.py to use pytest, create script and invoke it from Makefile
* simplify test_validate_cidr() using pytest.mark.parametrize
* simplify some tests using pytest.mark.parametrize
* finish applying pytest.mark.parametrize
* clean up Makefile test-python
* add blank line to hack/unit-test-python.sh
* fix typo in test case
* fix mega-linter error, blank space
* fix test case to fail due to invalid range
* fix typo in beeing to be being
* remove redundant test case
* reformat code for better readability
* add missing license to __init__.py files

Makefile

@@ -181,7 +181,12 @@ lint-admin-portal:
 test-python: pyenv az
 	. pyenv/bin/activate && \
 		azdev linter && \
-		azdev style 
+		azdev style && \
+		hack/unit-test-python.sh
+
+
+unit-test-python:
+	hack/unit-test-python.sh
 
 admin.kubeconfig:
 	hack/get-admin-kubeconfig.sh /subscriptions/${AZURE_SUBSCRIPTION_ID}/resourceGroups/${RESOURCEGROUP}/providers/Microsoft.RedHatOpenShift/openShiftClusters/${CLUSTER} >admin.kubeconfig

hack/unit-test-python.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+
+# This script executes Python unit tests. It first tries to activate the virtual environment, 
+# but if it does not exist it will be created and then activated.
+# When executing "pytest" command, it ignores integration test folder as 
+# this script is intended to just execute unit tests. 
+
+. pyenv/bin/activate || python3 -m venv pyenv && . pyenv/bin/activate
+pip install pytest
+cd python
+pytest --ignore=az/aro/azext_aro/tests/latest/integration

python/az/aro/azext_aro/_validators.py

@@ -24,70 +24,76 @@
 def validate_cidr(key):
     def _validate_cidr(namespace):
         cidr = getattr(namespace, key)
-        if cidr is not None:
-            try:
-                ipaddress.IPv4Network(cidr)
-            except ValueError as e:
-                raise InvalidArgumentValueError(f"Invalid --{key.replace('_', '-')} '{cidr}'.") from e
+        if cidr is None:
+            return
+        try:
+            ipaddress.IPv4Network(cidr)
+        except ValueError as e:
+            raise InvalidArgumentValueError(f"Invalid --{key.replace('_', '-')} '{cidr}'.") from e
 
     return _validate_cidr
 
 
 def validate_client_id(namespace):
-    if namespace.client_id is not None:
-        try:
-            uuid.UUID(namespace.client_id)
-        except ValueError as e:
-            raise InvalidArgumentValueError(f"Invalid --client-id '{namespace.client_id}'.") from e
+    if namespace.client_id is None:
+        return
+    try:
+        uuid.UUID(namespace.client_id)
+    except ValueError as e:
+        raise InvalidArgumentValueError(f"Invalid --client-id '{namespace.client_id}'.") from e
 
-        if namespace.client_secret is None or not str(namespace.client_secret):
-            raise RequiredArgumentMissingError('Must specify --client-secret with --client-id.')
+    if namespace.client_secret is None or not str(namespace.client_secret):
+        raise RequiredArgumentMissingError('Must specify --client-secret with --client-id.')
 
 
 def validate_client_secret(isCreate):
     def _validate_client_secret(namespace):
-        if isCreate and namespace.client_secret is not None:
-            if namespace.client_id is None or not str(namespace.client_id):
-                raise RequiredArgumentMissingError('Must specify --client-id with --client-secret.')
+        if not isCreate or namespace.client_secret is None:
+            return
+        if namespace.client_id is None or not str(namespace.client_id):
+            raise RequiredArgumentMissingError('Must specify --client-id with --client-secret.')
 
     return _validate_client_secret
 
 
 def validate_cluster_resource_group(cmd, namespace):
-    if namespace.cluster_resource_group is not None:
-        client = get_mgmt_service_client(
-            cmd.cli_ctx, ResourceType.MGMT_RESOURCE_RESOURCES)
+    if namespace.cluster_resource_group is None:
+        return
+    client = get_mgmt_service_client(
+        cmd.cli_ctx, ResourceType.MGMT_RESOURCE_RESOURCES)
 
-        if client.resource_groups.check_existence(namespace.cluster_resource_group):
-            raise InvalidArgumentValueError(
-                f"Invalid --cluster-resource-group '{namespace.cluster_resource_group}':"
-                " resource group must not exist.")
+    if client.resource_groups.check_existence(namespace.cluster_resource_group):
+        raise InvalidArgumentValueError(
+            f"Invalid --cluster-resource-group '{namespace.cluster_resource_group}':"
+            " resource group must not exist.")
 
 
 def validate_disk_encryption_set(cmd, namespace):
-    if namespace.disk_encryption_set is not None:
-        if not is_valid_resource_id(namespace.disk_encryption_set):
-            raise InvalidArgumentValueError(
-                f"Invalid --disk-encryption-set '{namespace.disk_encryption_set}', has to be a resource ID.")
-
-        desid = parse_resource_id(namespace.disk_encryption_set)
-        compute_client = get_mgmt_service_client(cmd.cli_ctx, ResourceType.MGMT_COMPUTE)
-        try:
-            compute_client.disk_encryption_sets.get(resource_group_name=desid['resource_group'],
-                                                    disk_encryption_set_name=desid['name'])
-        except CloudError as err:
-            raise InvalidArgumentValueError(
-                f"Invald --disk-encryption-set, error when getting '{namespace.disk_encryption_set}':"
-                f" {str(err)}") from err
+    if namespace.disk_encryption_set is None:
+        return
+    if not is_valid_resource_id(namespace.disk_encryption_set):
+        raise InvalidArgumentValueError(
+            f"Invalid --disk-encryption-set '{namespace.disk_encryption_set}', has to be a resource ID.")
+
+    desid = parse_resource_id(namespace.disk_encryption_set)
+    compute_client = get_mgmt_service_client(cmd.cli_ctx, ResourceType.MGMT_COMPUTE)
+    try:
+        compute_client.disk_encryption_sets.get(resource_group_name=desid['resource_group'],
+                                                disk_encryption_set_name=desid['name'])
+    except CloudError as err:
+        raise InvalidArgumentValueError(
+            f"Invald --disk-encryption-set, error when getting '{namespace.disk_encryption_set}':"
+            f" {str(err)}") from err
 
 
 def validate_domain(namespace):
-    if namespace.domain is not None:
-        if not re.match(r'^' +
-                        r'([a-z0-9]|[a-z0-9][-a-z0-9]{0,61}[a-z0-9])' +
-                        r'(\.([a-z0-9]|[a-z0-9][-a-z0-9]{0,61}[a-z0-9]))*' +
-                        r'$', namespace.domain):
-            raise InvalidArgumentValueError(f"Invalid --domain '{namespace.domain}'.")
+    if namespace.domain is None:
+        return
+    if not re.match(r'^' +
+                    r'([a-z0-9]|[a-z0-9][-a-z0-9]{0,61}[a-z0-9])' +
+                    r'(\.([a-z0-9]|[a-z0-9][-a-z0-9]{0,61}[a-z0-9]))*' +
+                    r'$', namespace.domain):
+        raise InvalidArgumentValueError(f"Invalid --domain '{namespace.domain}'.")
 
 
 def validate_pull_secret(namespace):
@@ -97,20 +103,23 @@ def validate_pull_secret(namespace):
             "Red Hat or from certified partners."
 
         logger.warning(warning)
+        return
 
-    else:
-        try:
-            if not isinstance(json.loads(namespace.pull_secret), dict):
-                raise Exception()
-        except Exception as e:
-            raise InvalidArgumentValueError("Invalid --pull-secret.") from e
+    try:
+        if not isinstance(json.loads(namespace.pull_secret), dict):
+            raise Exception()
+    except Exception as e:
+        raise InvalidArgumentValueError("Invalid --pull-secret.") from e
 
 
 def validate_sdn(namespace):
-    if namespace.software_defined_network is not None:
-        if namespace.software_defined_network not in ['OVNKubernetes', 'OpenshiftSDN']:
-            raise InvalidArgumentValueError(
-                f"Invalid --software-defined-network '{namespace.software_defined_network}'.")
+    if namespace.software_defined_network is None:
+        return
+
+    target_values = ['OVNKubernetes', 'OpenshiftSDN']
+    if namespace.software_defined_network not in target_values:
+        raise InvalidArgumentValueError(
+            f"Invalid --software-defined-network '{namespace.software_defined_network}'.")
 
 
 def validate_subnet(key):
@@ -132,15 +141,18 @@ def _validate_subnet(cmd, namespace):
             raise InvalidArgumentValueError(
                 f"--{key.replace('_', '-')} subscription '{parts['subscription']}' must equal cluster subscription.")
 
-        if parts['namespace'].lower() != 'microsoft.network':
+        expected_namespace = 'microsoft.network'
+        if parts['namespace'].lower() != expected_namespace:
             raise InvalidArgumentValueError(
                 f"--{key.replace('_', '-')} namespace '{parts['namespace']}' must equal Microsoft.Network.")
 
-        if parts['type'].lower() != 'virtualnetworks':
+        expected_type = 'virtualnetworks'
+        if parts['type'].lower() != expected_type:
             raise InvalidArgumentValueError(
                 f"--{key.replace('_', '-')} type '{parts['type']}' must equal virtualNetworks.")
 
-        if parts['last_child_num'] != 1:
+        expected_last_child_num = 1
+        if parts['last_child_num'] != expected_last_child_num:
             raise InvalidArgumentValueError(f"--{key.replace('_', '-')} '{subnet}' must have one child.")
 
         if 'child_namespace_1' in parts:
@@ -185,26 +197,30 @@ def validate_subnets(master_subnet, worker_subnet):
 def validate_visibility(key):
     def _validate_visibility(namespace):
         visibility = getattr(namespace, key)
-        if visibility is not None:
-            visibility = visibility.capitalize()
-            if visibility not in ['Private', 'Public']:
-                raise InvalidArgumentValueError(f"Invalid --{key.replace('_', '-')} '{visibility}'.")
+        if visibility is None:
+            return
+        visibility = visibility.capitalize()
+
+        possible_visibilities = ['Private', 'Public']
+        if visibility not in possible_visibilities:
+            raise InvalidArgumentValueError(f"Invalid --{key.replace('_', '-')} '{visibility}'.")
 
     return _validate_visibility
 
 
 def validate_vnet(cmd, namespace):
     validate_vnet_resource_group_name(namespace)
 
-    if namespace.vnet:
-        if not is_valid_resource_id(namespace.vnet):
-            namespace.vnet = resource_id(
-                subscription=get_subscription_id(cmd.cli_ctx),
-                resource_group=namespace.vnet_resource_group_name,
-                namespace='Microsoft.Network',
-                type='virtualNetworks',
-                name=namespace.vnet,
-            )
+    if not namespace.vnet:
+        return
+    if not is_valid_resource_id(namespace.vnet):
+        namespace.vnet = resource_id(
+            subscription=get_subscription_id(cmd.cli_ctx),
+            resource_group=namespace.vnet_resource_group_name,
+            namespace='Microsoft.Network',
+            type='virtualNetworks',
+            name=namespace.vnet,
+        )
 
 
 def validate_vnet_resource_group_name(namespace):
@@ -213,18 +229,27 @@ def validate_vnet_resource_group_name(namespace):
 
 
 def validate_worker_count(namespace):
-    if namespace.worker_count:
-        if namespace.worker_count < 3:
-            raise InvalidArgumentValueError('--worker-count must be greater than or equal to 3.')
+    if not namespace.worker_count:
+        return
+
+    minimum_workers_count = 3
+    if namespace.worker_count < minimum_workers_count:
+        raise InvalidArgumentValueError('--worker-count must be greater than or equal to ' + str(minimum_workers_count))
 
 
 def validate_worker_vm_disk_size_gb(namespace):
-    if namespace.worker_vm_disk_size_gb:
-        if namespace.worker_vm_disk_size_gb < 128:
-            raise InvalidArgumentValueError('--worker-vm-disk-size-gb must be greater than or equal to 128.')
+    if not namespace.worker_vm_disk_size_gb:
+        return
+
+    minimum_worker_vm_disk_size_gb = 128
+    if namespace.worker_vm_disk_size_gb < minimum_worker_vm_disk_size_gb:
+        error_msg = '--worker-vm-disk-size-gb must be greater than or equal to ' + str(minimum_worker_vm_disk_size_gb)
+
+        raise InvalidArgumentValueError(error_msg)
 
 
 def validate_refresh_cluster_credentials(namespace):
-    if namespace.refresh_cluster_credentials:
-        if namespace.client_secret is not None or namespace.client_id is not None:
-            raise RequiredArgumentMissingError('--client-id and --client-secret must be not set with --refresh-credentials.')  # pylint: disable=line-too-long
+    if not namespace.refresh_cluster_credentials:
+        return
+    if namespace.client_secret is not None or namespace.client_id is not None:
+        raise RequiredArgumentMissingError('--client-id and --client-secret must be not set with --refresh-credentials.')  # pylint: disable=line-too-long

python/az/aro/azext_aro/tests/latest/integration/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the Apache License 2.0.

python/az/aro/azext_aro/tests/latest/test_aro_scenario.py → python/az/aro/azext_aro/tests/latest/integration/test_aro_scenario.py

@@ -2,8 +2,7 @@
 # Licensed under the Apache License 2.0.
 
 import os
-import unittest
-import mock
+from unittest import mock
 
 from azure_devtools.scenario_tests import AllowLargeResponse
 from azure.cli.testsdk import ResourceGroupPreparer

python/az/aro/azext_aro/tests/latest/unit/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the Apache License 2.0.

python/az/aro/azext_aro/tests/latest/test_aro_helpers.py → python/az/aro/azext_aro/tests/latest/unit/test_aro_helpers.py

@@ -2,7 +2,7 @@
 # Licensed under the Apache License 2.0.
 
 import unittest
-import mock
+from unittest import mock
 from azext_aro.custom import generate_random_id