Merge pull request #29 from nutanixdev/version_bumps

chrisrnutanix · web-flow · commit a3ecd8fc62ad · 2025-01-20T11:15:52.000+11:00
Version bumps
diff --git a/python/v4api_sdk/batch_ops_actions.py b/python/v4api_sdk/batch_ops_actions.py
@@ -46,7 +46,7 @@
 from ntnx_prism_py_client.models.prism.v4.operations.ActionType import ActionType
 
 
-from tme import Utils
+from tme.utils import Utils
 
 
 def confirm_entity(api, client, entity_name: str) -> str:
diff --git a/python/v4api_sdk/batch_ops_create.py b/python/v4api_sdk/batch_ops_create.py
@@ -31,7 +31,7 @@
 
 from ntnx_prism_py_client.models.prism.v4.operations.ActionType import ActionType
 
-from tme import Utils
+from tme.utils import Utils
 
 
 def main():
diff --git a/python/v4api_sdk/batch_ops_modify.py b/python/v4api_sdk/batch_ops_modify.py
@@ -38,7 +38,7 @@
 
 from ntnx_prism_py_client.models.prism.v4.operations.ActionType import ActionType
 
-from tme import Utils
+from tme.utils import Utils
 
 
 def main():
diff --git a/python/v4api_sdk/create_categories.py b/python/v4api_sdk/create_categories.py
@@ -21,7 +21,7 @@
 from ntnx_prism_py_client.rest import ApiException as PrismException
 
 
-from tme import Utils
+from tme.utils import Utils
 from tme import Config
 
 
diff --git a/python/v4api_sdk/create_image_sdk.py b/python/v4api_sdk/create_image_sdk.py
@@ -18,7 +18,7 @@
 from ntnx_clustermgmt_py_client import ApiClient as ClusterClient
 from ntnx_clustermgmt_py_client import Configuration as ClusterConfiguration
 
-from tme import Utils
+from tme.utils import Utils
 
 
 def main():
@@ -46,8 +46,7 @@ def main():
 
     # get the cluster password
     cluster_password = getpass.getpass(
-            prompt="Please enter your Prism Central \
-                    password: ",
+            prompt="Please enter your Prism Central password: ",
                     stream=None,
                     )
 
diff --git a/python/v4api_sdk/create_network_security_policy.py b/python/v4api_sdk/create_network_security_policy.py
@@ -30,7 +30,7 @@
 from ntnx_prism_py_client.rest import ApiException as PrismException
 
 
-from tme import Utils
+from tme.utils import Utils
 from tme import Config
 
 
diff --git a/python/v4api_sdk/create_subnet_sdk.py b/python/v4api_sdk/create_subnet_sdk.py
@@ -36,8 +36,8 @@
 from ntnx_prism_py_client.rest import ApiException as PrismException
 
 
-from tme import Utils
-from tme import Config
+from tme.utils import Utils
+from tme.utils import Config
 
 
 def confirm_entity(api, client, entity_name: str, exclusions: list) -> str:
@@ -297,7 +297,7 @@ def main():
 
     # did the user say Yes to creating the subnet?
     if confirm_create:
-        networking_instance = ntnx_networking_py_client.api.SubnetApi(
+        networking_instance = ntnx_networking_py_client.api.SubnetsApi(
             api_client=networking_client
         )
 
diff --git a/python/v4api_sdk/create_vm_sdk.py b/python/v4api_sdk/create_vm_sdk.py
@@ -29,7 +29,7 @@
 from ntnx_networking_py_client import Configuration as NetworkingConfiguration
 from ntnx_networking_py_client import ApiClient as NetworkingClient
 
-from tme import Utils
+from tme.utils import Utils
 
 
 def confirm_entity(api, client, entity_name: str, exclusions: list) -> str:
diff --git a/python/v4api_sdk/dr/.gitignore b/python/v4api_sdk/dr/.gitignore
@@ -0,0 +1,2 @@
+.DS_Store
+venv/
diff --git a/python/v4api_sdk/requirements.txt b/python/v4api_sdk/requirements.txt
@@ -1,10 +1,10 @@
 requests==2.31.0
 ruff==0.1.15
-ntnx_vmm_py_client==4.0.1b1
-ntnx_lifecycle_py_client==4.0.1b1
-ntnx_prism_py_client==4.0.1b1
-ntnx_clustermgmt_py_client==4.0.1b2
-ntnx_aiops_py_client==4.0.3a2
-ntnx_networking_py_client==4.0.1b1
-ntnx_microseg_py_client==4.0.1b1
-ntnx_iam_py_client==4.0.1b1
+ntnx_vmm_py_client==4.0.1
+ntnx_lifecycle_py_client==4.0.1
+ntnx_prism_py_client==4.0.1
+ntnx_clustermgmt_py_client==4.0.1
+ntnx_aiops_py_client==4.0.1
+ntnx_networking_py_client==4.0.1
+ntnx_microseg_py_client==4.0.1
+ntnx_iam_py_client==4.0.1
diff --git a/python/v4api_sdk/tme/__init__.py b/python/v4api_sdk/tme/__init__.py
@@ -1,114 +0,0 @@
-"""
-Simple module to allow function re-use across Nutanix
-v4 SDK code samples
-
-Requires Prism Central 2024.1 or later, AOS 6.8 or later
-"""
-
-import time
-import urllib3
-from dataclasses import dataclass
-from timeit import default_timer as timer
-
-import ntnx_prism_py_client
-from ntnx_prism_py_client import ApiClient as PrismClient
-from ntnx_prism_py_client import Configuration as PrismConfiguration
-
-
-@dataclass
-class Config:
-    """
-    dataclass to hold configuration for each script run
-    nice and modular
-    """
-
-    pc_ip: str
-    pc_username: str
-    pc_password: str
-
-class Utils:
-    """
-    class to manage simple reusable functions across the Python
-    v4 API code samples
-    """
-
-    prism_config: PrismConfiguration
-
-    def __init__(self, pc_ip: str, username: str, password: str):
-        """
-        class constructor
-        create reusable instances of Prism connections (etc)
-        """
-        self.prism_config = PrismConfiguration()
-        self.prism_config.host = pc_ip
-        self.prism_config.username = username
-        self.prism_config.password = password
-        self.prism_config.verify_ssl = False
-        self.prism_client = PrismClient(configuration=self.prism_config)
-        self.prism_client.add_default_header(
-            header_name="Accept-Encoding", header_value="gzip, deflate, br"
-        )
-        self.prism_instance = ntnx_prism_py_client.api.TasksApi(
-            api_client=self.prism_client
-        )
-        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
-
-    def get_task(self, ext_id: str):
-        """
-        method used to get details of a specified task
-        """
-        task = self.prism_instance.get_task_by_id(f"{ext_id}")
-        if task:
-            return task
-        return None
-
-    def confirm(self, message: str):
-        """
-        method to request a yes/NO confirmation from the user
-        used to run or skip precheck, inventory (etc)
-        """
-        yes_no = input(f"{message} (yes/NO): ").lower()
-        return yes_no == "yes"
-
-    def monitor_task(
-        self, task_ext_id, task_name, pc_ip, username, password, poll_timeout, prefix = ""
-    ):
-        """
-        method used to monitor Prism Central tasks
-        will print a series of period characters and re-check task
-        status at the specified interval
-        this version uses the Prism SDK
-        """
-        start = timer()
-        # print message until specified  task is finished
-        prism_config = PrismConfiguration()
-        prism_config.host = pc_ip
-        prism_config.username = username
-        prism_config.password = password
-        prism_config.verify_ssl = False
-        prism_client = PrismClient(configuration=prism_config)
-        prism_client.add_default_header(
-            header_name="Accept-Encoding", header_value="gzip, deflate, br"
-        )
-        prism_instance = ntnx_prism_py_client.api.TasksApi(api_client=prism_client)
-        task = prism_instance.get_task_by_id(f"{prefix}{task_ext_id}")
-        units = "second" if poll_timeout == 1 else "seconds"
-        print(
-            f"{task_name} running, checking progress every {poll_timeout} {units} ...",
-            end="",
-        )
-        while True:
-            if task.data.status == "RUNNING":
-                print(".", end="", flush=True)
-            else:
-                print(" finished.")
-                break
-            time.sleep(int(poll_timeout))
-            task = prism_instance.get_task_by_id(f"{prefix}{task_ext_id}")
-        end = timer()
-        elapsed_time = end - start
-        if elapsed_time <= 60:
-            duration = f"{round(elapsed_time, 0)} seconds"
-        else:
-            duration = f"{round(elapsed_time // 60, 0)} minutes"
-        return duration
diff --git a/python/v4api_sdk/tme/test.py b/python/v4api_sdk/tme/test.py
@@ -0,0 +1,15 @@
+import unittest
+from utils import Utils
+
+class TestUtils(unittest.TestCase):
+
+    def test_confirm_yes(self):
+        utils = Utils()
+        self.assertEqual(utils.confirm("Enter yes:"), True)
+
+    def test_confirm_no(self):
+        utils = Utils()
+        self.assertEqual(utils.confirm("Enter no:"), False)
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/python/v4api_sdk/tme/utils.py b/python/v4api_sdk/tme/utils.py
@@ -0,0 +1,118 @@
+"""
+Simple module to allow function re-use across Nutanix
+v4 SDK code samples
+
+Requires Prism Central 2024.1 or later, AOS 6.8 or later
+"""
+
+import time
+import urllib3
+from dataclasses import dataclass
+from timeit import default_timer as timer
+
+import ntnx_prism_py_client
+from ntnx_prism_py_client import ApiClient as PrismClient
+from ntnx_prism_py_client import Configuration as PrismConfiguration
+
+
+@dataclass
+class Config:
+    """
+    dataclass to hold configuration for each script run
+    nice and modular
+    """
+
+    pc_ip: str
+    pc_username: str
+    pc_password: str
+
+class Utils:
+    """
+    class to manage simple reusable functions across the Python
+    v4 API code samples
+    """
+
+    prism_config: PrismConfiguration
+
+    def __init__(self, pc_ip: str, username: str, password: str):
+        """
+        class constructor
+        create reusable instances of Prism connections (etc)
+        """
+        self.prism_config = PrismConfiguration()
+        self.prism_config.host = pc_ip
+        self.prism_config.username = username
+        self.prism_config.password = password
+        self.prism_config.verify_ssl = False
+        self.prism_client = PrismClient(configuration=self.prism_config)
+        self.prism_client.add_default_header(
+            header_name="Accept-Encoding", header_value="gzip, deflate, br"
+        )
+        self.prism_instance = ntnx_prism_py_client.api.TasksApi(
+            api_client=self.prism_client
+        )
+        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+    def get_task(self, ext_id: str):
+        """
+        method used to get details of a specified task
+        """
+        task = self.prism_instance.get_task_by_id(f"{ext_id}")
+        if task:
+            return task
+        return None
+
+    def confirm(self, message: str):
+        """
+        method to request a yes/NO confirmation from the user
+        used to run or skip precheck, inventory (etc)
+        """
+        yes_no = input(f"{message} (yes/NO): ").lower()
+        return yes_no == "yes"
+
+    def monitor_task(
+        self, task_ext_id, task_name, pc_ip, username, password, poll_timeout, prefix = ""
+    ):
+        """
+        method used to monitor Prism Central tasks
+        will print a series of period characters and re-check task
+        status at the specified interval
+        this version uses the Prism SDK
+        """
+        start = timer()
+        # print message until specified  task is finished
+        prism_config = PrismConfiguration()
+        prism_config.host = pc_ip
+        prism_config.username = username
+        prism_config.password = password
+        prism_config.verify_ssl = False
+        prism_client = PrismClient(configuration=prism_config)
+        prism_client.add_default_header(
+            header_name="Accept-Encoding", header_value="gzip, deflate, br"
+        )
+        prism_instance = ntnx_prism_py_client.api.TasksApi(api_client=prism_client)
+        task = prism_instance.get_task_by_id(f"{prefix}{task_ext_id}")
+        units = "second" if poll_timeout == 1 else "seconds"
+        print(
+            f"{task_name} running, checking progress every {poll_timeout} {units} (progress will update when percentage complete changes) ...",
+            end="",
+        )
+        percent_complete = 0
+        print(f" {task.data.progress_percentage}% ... ", end="", flush=True)
+        while True:
+            if task.data.status == "RUNNING":
+                if task.data.progress_percentage > percent_complete:
+                    print(f" {task.data.progress_percentage}% ... ", end="", flush=True)
+                    percent_complete = task.data.progress_percentage
+            else:
+                print("finished.")
+                break
+            time.sleep(int(poll_timeout))
+            task = prism_instance.get_task_by_id(f"{prefix}{task_ext_id}")
+        end = timer()
+        elapsed_time = end - start
+        if elapsed_time <= 60:
+            duration = f"{round(elapsed_time, 0)} seconds"
+        else:
+            duration = f"{round(elapsed_time // 60, 0)} minutes"
+        return duration
diff --git a/python/v4api_sdk/update_image_etag_sdk.py b/python/v4api_sdk/update_image_etag_sdk.py
@@ -17,7 +17,7 @@
 from ntnx_prism_py_client import ApiClient as PrismClient
 from ntnx_prism_py_client import Configuration as PrismConfiguration
 
-from tme import Utils
+from tme.utils import Utils
 
 """
 suppress warnings about insecure connections
