NSOL-6228: fix: replace az CLI auth with DefaultAzureCredential and clean up ANF/GCNV modules

sushma-m1 · sushma-m1 · commit 1baa863b46d0 · 2026-04-02T17:01:18.000+05:30
- client.py: swap AzureCliCredential + az subprocess for
  DefaultAzureCredential + SubscriptionClient; fix ANFClientManager
  cache invalidation (track _subscription_id, log on re-init)
- replication_management.py: remove DEFAULT_SERVICE_LEVEL; resolve
  service level from pool when not provided; deduplicate throughput
  calculation via _calculate_min_throughput_mibps() in base.py
- volume_management.py: replace inline throughput calc with
  _calculate_min_throughput_mibps() from base.py
- base.py, config.py, snapshot_management.py: add trailing newlines;
  raise InvalidConfigError instead of bare Exception
- netapp_dataops_anf_mcp.py: remove StandardZRS from service level
  docstrings; update destination_service_level description
- setup.cfg: add azure-mgmt-resource-subscriptions to azure extras
- setup_gcnv_auth.py: remove unnecessary f-strings and stray blank line
diff --git a/netapp_dataops_traditional/netapp_dataops/mcp_server/netapp_dataops_anf_mcp.py b/netapp_dataops_traditional/netapp_dataops/mcp_server/netapp_dataops_anf_mcp.py
@@ -86,7 +86,7 @@ async def create_volume_tool(
             Optional. The name of a delegated Azure subnet to construct the Azure Resource ID for a delegated subnet.
             Will use config default if not provided, otherwise defaults to "default".
         service_level (str):
-            Optional. Service level (Standard, Premium, Ultra, StandardZRS, Flexible).
+            Optional. Service level (Standard, Premium, Ultra, Flexible).
             Defaults to Premium.
         tags (Dict[str, str]):
             Optional. Resource tags.
@@ -303,7 +303,7 @@ async def clone_volume_tool(
             Optional. The name of a delegated Azure subnet to construct the Azure Resource ID for a delegated subnet.
             Will use config default if not provided, otherwise defaults to "default".
         service_level (str):
-            Optional. Service level (Standard, Premium, Ultra, StandardZRS, Flexible).
+            Optional. Service level (Standard, Premium, Ultra, Flexible).
             Defaults to Premium.
         protocol_types (List[str]):
             Optional. List of protocol types (NFSv3, NFSv4.1, CIFS). Will use config default if not provided.
@@ -680,8 +680,9 @@ async def create_replication_tool(
         destination_subnet_name (str):
             Optional. Destination subnet name. Defaults to "default".
         destination_service_level (str):
-            Optional. Destination service level (Standard/Premium/Ultra).
-            Defaults to "Premium".
+            Optional. Destination service level (Standard/Premium/Ultra/Flexible).
+            If not provided, it will be retrieved from the destination capacity pool.
+            Must be one of: Standard, Premium, Ultra, Flexible.
         destination_zones (list):
             Optional. Availability zones for destination volume. If None, defaults to ["1"].
         
diff --git a/netapp_dataops_traditional/netapp_dataops/traditional/anf/__init__.py b/netapp_dataops_traditional/netapp_dataops/traditional/anf/__init__.py
@@ -23,4 +23,4 @@
     "list_snapshots",
     "create_replication",
     "create_anf_config"
-]
+]
diff --git a/netapp_dataops_traditional/netapp_dataops/traditional/anf/base.py b/netapp_dataops_traditional/netapp_dataops/traditional/anf/base.py
@@ -7,6 +7,30 @@
 from typing import Any, Dict
 
 
+def _calculate_min_throughput_mibps(service_level: str, usage_threshold_bytes: int) -> float:
+    """
+    Calculate the minimum throughput in MiB/s for a manual QoS volume based on
+    service level and size.
+
+    Args:
+        service_level (str): One of Standard, Premium, Ultra, Flexible.
+        usage_threshold_bytes (int): Volume quota in bytes.
+
+    Returns:
+        float: Minimum throughput in MiB/s.
+    """
+    size_gib = usage_threshold_bytes / (1024 * 1024 * 1024)
+    level = service_level.lower()
+    if level == "standard":
+        return max(16.0, size_gib * 0.016)
+    elif level == "premium":
+        return max(64.0, size_gib * 0.064)
+    elif level == "ultra":
+        return max(128.0, size_gib * 0.128)
+    else:
+        return 64.0
+
+
 def _validate_required_params(**params) -> None:
     """Validate that all required parameters are provided.
     
@@ -116,4 +140,4 @@ def _get_clean_error_message(exception: Exception) -> str:
         return str(msg)
     
     # Fallback to full string representation for other exceptions
-    return error_str
+    return error_str
diff --git a/netapp_dataops_traditional/netapp_dataops/traditional/anf/client.py b/netapp_dataops_traditional/netapp_dataops/traditional/anf/client.py
@@ -1,93 +1,96 @@
 """
 NetApp DataOps Toolkit - Azure NetApp Files (ANF) Client Management
 This module handles Azure client authentication and management for ANF operations.
+
+Authentication is handled via DefaultAzureCredential, which automatically chains
+through the following methods (in order) without requiring environment variables:
+  1. Managed Identity (production: Azure VMs, AKS, App Service, etc.)
+  2. Workload Identity (Kubernetes with Azure AD Workload Identity)
+  3. Azure CLI ('az login' for local development)
+  4. Azure Developer CLI, VS Code, PowerShell, and others
+
+No secrets or environment variables are required. For local development, run:
+  az login --tenant <TENANT_ID>
+  az account set --subscription <SUBSCRIPTION_ID>
 """
 
 from typing import Tuple, Optional
-from azure.identity import AzureCliCredential
+from azure.identity import DefaultAzureCredential
 from azure.mgmt.netapp import NetAppManagementClient
+from azure.mgmt.resource.subscriptions import SubscriptionClient
 from azure.core.exceptions import ClientAuthenticationError
 
 from netapp_dataops.logging_utils import setup_logger
 
 logger = setup_logger(__name__)
 
+
 class ANFClientManager:
     """Singleton class for managing Azure NetApp Files client connections."""
-    
+
     _instance: Optional['ANFClientManager'] = None
     _client: Optional[NetAppManagementClient] = None
-    
+    _subscription_id: Optional[str] = None
+
     def __new__(cls) -> 'ANFClientManager':
         if cls._instance is None:
             cls._instance = super().__new__(cls)
         return cls._instance
-    
-    def get_client(self, subscription_id: str, print_output: Optional[bool] = False) -> NetAppManagementClient:
+
+    def get_client(self, subscription_id: str, credential: DefaultAzureCredential, print_output: Optional[bool] = False) -> NetAppManagementClient:
         """
-        Get or create an authenticated NetApp Management Client.
+        Get or create an authenticated NetApp Management Client using DefaultAzureCredential.
+        Re-creates the client if the subscription ID has changed.
         """
-        if self._client is None or self._client._config.subscription_id != subscription_id:
-            # Use AzureCliCredential to specifically respect 'az login' and 'az account set'
-            credential = AzureCliCredential()
+        if self._client is None or self._subscription_id != subscription_id:
+            if self._client is not None and self._subscription_id != subscription_id:
+                if print_output:
+                    logger.info(f"Subscription changed from '{self._subscription_id}' to '{subscription_id}'. Re-initializing ANF client.")
             try:
                 self._client = NetAppManagementClient(credential, subscription_id)
+                self._subscription_id = subscription_id
             except Exception as e:
                 raise ClientAuthenticationError(f"Failed to initialize ANF client: {str(e)}")
-        
+
         return self._client
 
 
 def get_anf_client(print_output: bool = False) -> Tuple[NetAppManagementClient, str]:
     """
     Convenience function to get an authenticated ANF client.
-    Automatically fetches the active Subscription ID from the 'az account show' context.
+
+    Uses DefaultAzureCredential for authentication and SubscriptionClient to
+    resolve the active subscription ID — no dependency on the az CLI at runtime.
     """
-    import subprocess
-    import json
-    
     try:
-        # Get the currently active subscription from Azure CLI
-        # This respects 'az account set --subscription <id>'
-        result = subprocess.run(
-            ['az', 'account', 'show'],
-            capture_output=True,
-            text=True,
-            check=True
-        )
-        
-        account_info = json.loads(result.stdout)
+        credential = DefaultAzureCredential()
+
+        # Resolve the active subscription via the Azure SDK (no az CLI required)
+        subscription_client = SubscriptionClient(credential)
+        subscriptions = list(subscription_client.subscriptions.list())
 
-        if 'id' not in account_info:
+        if not subscriptions:
             raise ClientAuthenticationError(
-                "Azure CLI output did not contain a subscription 'id'. "
-                "Please run: az login --tenant <TENANT_ID>"
+                "No Azure subscriptions found for the authenticated identity. "
+                "Ensure the credential has access to at least one subscription."
             )
 
-        subscription_id = account_info['id']
-        
+        # Use the first available subscription (respects az account set via CLI credential)
+        subscription = subscriptions[0]
+        subscription_id = subscription.subscription_id
+
         if print_output:
-            subscription_name = account_info.get('name', 'unknown')
-            logger.info(f"Connected to ANF via Active Subscription: {subscription_name} ({subscription_id})")
-    
-    except subprocess.CalledProcessError as e:
-        logger.error(f"Azure CLI returned error: {e.stderr}")
-        raise ClientAuthenticationError(
-            f"Failed to get active subscription from Azure CLI: {e.stderr}\n"
-            "Please run: az login --tenant <TENANT_ID>"
-        )
-    except json.JSONDecodeError as e:
-        logger.error(f"Azure CLI output not valid JSON: {str(e)}")
-        raise ClientAuthenticationError(f"Failed to parse Azure CLI output: {str(e)}")
-    except FileNotFoundError:
-        logger.error("Azure CLI (az) not found in PATH")
+            logger.info(f"Connected to ANF via Active Subscription: {subscription.display_name} ({subscription_id})")
+
+    except ClientAuthenticationError:
+        raise
+    except Exception as e:
         raise ClientAuthenticationError(
-            "Azure CLI (az) not found. Please install it from: "
-            "https://docs.microsoft.com/en-us/cli/azure/install-azure-cli"
+            f"Failed to authenticate with Azure: {str(e)}\n"
+            "Please run: az login or az login --tenant <TENANT_ID>\n"
         )
-    
-    # Use AzureCliCredential with the detected subscription
     manager = ANFClientManager()
-    client = manager.get_client(subscription_id, print_output=print_output)
-    
+    client = manager.get_client(subscription_id, credential, print_output=print_output)
+
     return client, subscription_id
+
diff --git a/netapp_dataops_traditional/netapp_dataops/traditional/anf/config.py b/netapp_dataops_traditional/netapp_dataops/traditional/anf/config.py
@@ -447,7 +447,7 @@ def _get_service_level_from_pool(
         str: The service level of the capacity pool (Standard, Premium, or Ultra).
         
     Raises:
-        Exception: If there's an error retrieving the capacity pool information.
+        InvalidConfigError: If there's an error retrieving the capacity pool information.
     """
     try:
         # Import here to avoid circular dependency
@@ -480,4 +480,4 @@ def _get_service_level_from_pool(
         error_msg = f"Failed to retrieve service level from capacity pool '{pool_name}': {str(e)}"
         if print_output:
             logger.error(error_msg)
-        raise Exception(error_msg)
+        raise InvalidConfigError(error_msg)
diff --git a/netapp_dataops_traditional/netapp_dataops/traditional/anf/replication_management.py b/netapp_dataops_traditional/netapp_dataops/traditional/anf/replication_management.py
@@ -7,8 +7,8 @@
 from typing import Dict, List, Any
 from azure.mgmt.netapp.models import AuthorizeRequest
 from azure.core.exceptions import ResourceNotFoundError
-from .client import get_anf_client
-from .base import _serialize, _validate_required_params, _get_clean_error_message
+from .client import _get_anf_client
+from .base import _serialize, _validate_required_params, _get_clean_error_message, _calculate_min_throughput_mibps
 from .config import _retrieve_anf_config, _get_config_value, InvalidConfigError
 
 from netapp_dataops.logging_utils import setup_logger
@@ -18,7 +18,6 @@
 # Constants for validation
 VALID_PROTOCOL_TYPES = ["NFSv3", "NFSv4.1", "CIFS"]
 VALID_SERVICE_LEVELS = ["Standard", "Premium", "Ultra", "Flexible"]
-DEFAULT_SERVICE_LEVEL = "Premium"
 
 def create_replication(
     # Source volume parameters
@@ -78,8 +77,9 @@ def create_replication(
         destination_subnet_name (str):
             Optional. Destination subnet name. Defaults to "default".
         destination_service_level (str):
-            Optional. Destination service level (Standard/Premium/Ultra).
-            Defaults to "Premium".
+            Optional. Destination service level (Standard/Premium/Ultra/Flexible).
+            If not provided, it will be retrieved from the destination capacity pool.
+            Must be one of: Standard, Premium, Ultra, Flexible.
         destination_zones (List[str]):
             Optional. Availability zones for destination volume. If None, defaults to ["1"].
         destination_throughput_mibps (float):
@@ -145,23 +145,13 @@ def create_replication(
             destination_virtual_network_name=destination_virtual_network_name
         )
         
-        # Set default service level if not provided
-        if destination_service_level is None:
-            destination_service_level = DEFAULT_SERVICE_LEVEL
-        
         # Validate protocol types
         invalid_protocols = [pt for pt in destination_protocol_types if pt not in VALID_PROTOCOL_TYPES]
         if invalid_protocols:
             error_message = f"Invalid protocol types: {invalid_protocols}. Valid options are: {VALID_PROTOCOL_TYPES}"
             logger.error(error_message)
             return {"status": "error", "details": error_message}
         
-        # Validate service level
-        if destination_service_level not in VALID_SERVICE_LEVELS:
-            error_message = f"Invalid service level: '{destination_service_level}'. Valid options are: {VALID_SERVICE_LEVELS}"
-            logger.error(error_message)
-            return {"status": "error", "details": error_message}
-        
         # Get ANF client and subscription ID (automatically retrieved from Azure CLI)
         client, subscription_id = get_anf_client(print_output=print_output)
 
@@ -195,12 +185,22 @@ def create_replication(
                 )
 
                 # If service level not provided, get it from destination pool
-                if destination_service_level == DEFAULT_SERVICE_LEVEL:
+                if destination_service_level is None:
                     if hasattr(destination_pool, 'service_level') and destination_pool.service_level:
-                        destination_service_level = destination_pool.service_level
+                        destination_service_level = str(destination_pool.service_level).split('.')[-1].capitalize()
                         if print_output:
                             logger.info(f"Using service level from destination pool: {destination_service_level}")
 
+                # Validate service level after pool-lookup override
+                if destination_service_level is None:
+                    error_message = "destination_service_level is required and could not be retrieved from the destination pool"
+                    logger.error(error_message)
+                    return {"status": "error", "details": error_message}
+                if destination_service_level not in VALID_SERVICE_LEVELS:
+                    error_message = f"Invalid service level: '{destination_service_level}'. Valid options are: {VALID_SERVICE_LEVELS}"
+                    logger.error(error_message)
+                    return {"status": "error", "details": error_message}
+
                 destination_qos_type = destination_pool.qos_type
                 
                 # Extract QoS type value (e.g., "manual" from "QosType.MANUAL")
@@ -227,30 +227,16 @@ def create_replication(
                                 logger.info(f"Using throughput_mibps from source volume: {destination_throughput_mibps}")
                         else:
                             # Calculate minimum throughput based on service level and size
-                            size_gib = destination_usage_threshold / (1024 * 1024 * 1024)
-                            if destination_service_level.lower() == "standard":
-                                destination_throughput_mibps = max(16.0, size_gib * 0.016)
-                            elif destination_service_level.lower() == "premium":
-                                destination_throughput_mibps = max(64.0, size_gib * 0.064)
-                            elif destination_service_level.lower() == "ultra":
-                                destination_throughput_mibps = max(128.0, size_gib * 0.128)
-                            else:
-                                destination_throughput_mibps = 64.0
-                            
+                            destination_throughput_mibps = _calculate_min_throughput_mibps(
+                                destination_service_level, destination_usage_threshold
+                            )
                             if print_output:
                                 logger.info(f"Calculated destination throughput_mibps for manual QoS: {destination_throughput_mibps}")
                     except Exception as e:
                         # If we can't get source volume info, calculate based on destination size
-                        size_gib = destination_usage_threshold / (1024 * 1024 * 1024)
-                        if destination_service_level.lower() == "standard":
-                            destination_throughput_mibps = max(16.0, size_gib * 0.016)
-                        elif destination_service_level.lower() == "premium":
-                            destination_throughput_mibps = max(64.0, size_gib * 0.064)
-                        elif destination_service_level.lower() == "ultra":
-                            destination_throughput_mibps = max(128.0, size_gib * 0.128)
-                        else:
-                            destination_throughput_mibps = 64.0
-                        
+                        destination_throughput_mibps = _calculate_min_throughput_mibps(
+                            destination_service_level, destination_usage_threshold
+                        )
                         if print_output:
                             logger.info(f"Calculated destination throughput_mibps for manual QoS (source unavailable): {destination_throughput_mibps}")
                 
@@ -266,16 +252,9 @@ def create_replication(
                 # If we can't get pool info and throughput is still None, set a safe default
                 # This is important for manual QoS pools
                 if destination_throughput_mibps is None:
-                    size_gib = destination_usage_threshold / (1024 * 1024 * 1024)
-                    if destination_service_level.lower() == "standard":
-                        destination_throughput_mibps = max(16.0, size_gib * 0.016)
-                    elif destination_service_level.lower() == "premium":
-                        destination_throughput_mibps = max(64.0, size_gib * 0.064)
-                    elif destination_service_level.lower() == "ultra":
-                        destination_throughput_mibps = max(128.0, size_gib * 0.128)
-                    else:
-                        destination_throughput_mibps = 64.0
-                    
+                    destination_throughput_mibps = _calculate_min_throughput_mibps(
+                        destination_service_level, destination_usage_threshold
+                    )
                     if print_output:
                         logger.info(f"Using calculated throughput_mibps (pool QoS type unavailable): {destination_throughput_mibps}")
             
diff --git a/netapp_dataops_traditional/netapp_dataops/traditional/anf/snapshot_management.py b/netapp_dataops_traditional/netapp_dataops/traditional/anf/snapshot_management.py
@@ -15,7 +15,6 @@
 
 logger = setup_logger(__name__)
 
-
 def create_snapshot(
     snapshot_name: str,
     volume_name: str,
diff --git a/netapp_dataops_traditional/netapp_dataops/traditional/anf/volume_management.py b/netapp_dataops_traditional/netapp_dataops/traditional/anf/volume_management.py
diff --git a/netapp_dataops_traditional/netapp_dataops/traditional/gcnv/setup_gcnv_auth.py b/netapp_dataops_traditional/netapp_dataops/traditional/gcnv/setup_gcnv_auth.py
diff --git a/netapp_dataops_traditional/setup.cfg b/netapp_dataops_traditional/setup.cfg
