hello
Server : Apache/2.4.52 (Ubuntu) System : Linux HAN2 5.15.0-185-generic #195-Ubuntu SMP Fri Jun 19 17:11:50 UTC 2026 x86_64 User : servadmin ( 1000) PHP Version : 8.1.2-1ubuntu2.25 Disable Function : NONE Directory : /usr/local/nutanix/ngt/ngtcli/ |
# Copyright (c) 2016 Nutanix Inc. All rights reserved.
#
# Author: annie.pillai@nutanix.com
#
# This class implements all utility functions required by the
# cli module.
#
import datetime
import json
import re
from ngtcli_consts import *
from ssr_utils.ssr_consts import *
def print_util_for_disks(disk_list, index, print_list):
'''
Util function to print rows for disk info.
'''
disk = disk_list[index]
print_list.append(disk.get(DISK_LABEL))
print_list.append(" ".join(disk.get(LOGICAL_DRIVE)))
def print_util_for_time(time_value, print_list):
'''
Util function to print time.
'''
pattern = '%Y.%m.%d %I:%M %p'
time_value = datetime.datetime.fromtimestamp(time_value).strftime(pattern)
print_list.append(time_value)
def print_row_for_snapshot(snap_id, disk_list, index, create_time):
'''
Util function to print row for snapshot info.
'''
print_list = [snap_id]
print_util_for_disks(disk_list, index, print_list)
print_util_for_time(create_time, print_list)
print(SNAPSHOT_TEMPLATE.format(*print_list))
def print_row_for_attached_disks(snap_id, original, attached, detach_time):
'''
Util function to print row for attached disk info.
'''
print_list = [snap_id, original, attached]
print_util_for_time(detach_time, print_list)
print(DISK_TEMPLATE.format(*print_list))
def print_capabilities(result):
capabilities = result.get("ngt_capabilities")
version = result.get("ngt_version")
vss = capabilities.get("vm_snapshot_service_capabilities_bitmap")
script = capabilities.get("script_execution_capabilities_bitmap")
ssr = capabilities.get("file_level_restore_capabilities_bitmap")
ipless_support = (capabilities.get("supported_communication_types_bitmap") >> 1) & 1
print(NGT_TEMPLATE.format(version, vss, script, ssr, ipless_support))
def print_snapshot_header():
'''
Print the header for snapshot info table.
'''
print(SNAPSHOT_TEMPLATE.format(*SNAP_HEADERS))
print(SNAPSHOT_TEMPLATE.format(*['-'*12, '-'*12, '-'*20, '-'*20]))
def print_disk_header():
'''
Print the header for attached disk info table.
'''
print(DISK_TEMPLATE.format(*DISK_HEADERS))
print(DISK_TEMPLATE.format(*['-'*12, '-'*20, '-'*20, '-'*20]))
def remove_unwanted_spaces(operation):
return re.sub(r'(\s)*(=)(\s)*', r'=', operation)
def parse_to_get_action_and_args(line):
'''
Parse the command to get the operation and the arguments.
'''
operation = remove_unwanted_spaces(line)
arguments = operation.split()
args_provided = []
if len(arguments) > 1:
args_provided = arguments[1:]
return arguments[0], args_provided
def exception_wrapped_func(func, *args):
'''
Wrapper to catch exceptions while executing a function.
'''
try:
func(*args)
except Exception as e:
if args[0]:
print_json(1, str(e))
else:
print(str(e))
return
def get_disk_details(result, label_key, disk_drive_key):
'''
Get info about disks.
Args:
result: response received from the backend.
label_key: Key for which info is needed.
disk_drive_key: Key to be displayed on the cli.
'''
disk_label = result.get(label_key)
disk_drives = " ".join(result.get(disk_drive_key))
details = "{0} ({1})".format(disk_label, disk_drives)
return details
def print_json(status, result_list):
'''
Util function to print in json format.
'''
response = {}
response["status"] = status
response["data"] = result_list
print(json.dumps(response))
def print_warnings(status_list):
'''
Util function to print warnings after attach disk operation
'''
for status in (status_list or []):
if status.get(DETAIL_STATUS) == kWARNING:
print("[WARNING] {} : {}".format(status.get(ATTACHED_DISK_PARTITION),
status.get(DETAIL_MESSAGE)))
def print_attached_disk(result):
'''
Util function to extract info from result and print values on cli
for an attached disk.
'''
detach_time = result.get(DISK_DETACH_TIME) / 1000
snap_id = result.get(SNAPSHOT_ID)
# Get original and attached disk details.
original = get_disk_details(result, ORIGINAL_DISK_LABEL,
ORIGINAL_DISK_DRIVE)
attached = get_disk_details(result, ATTACHED_DISK_LABEL,
ATTACHED_DISK_DRIVE)
# Print the row on the cli.
print_row_for_attached_disks(snap_id, original, attached, detach_time)
print_warnings(result.get(ATTACHED_DISK_DETAILS))
def print_for_attach_detach_op(is_json, generator_func, task, response,
completed_msg):
'''
Util function to print for attach or detach op.
'''
# For json formatted output.
if is_json:
result = [x for x in generator_func]
if "Completed" in result:
print_json(task.status, task.response or response)
return
# Print messages for each step.
for value in generator_func:
if value == "Completed":
print(completed_msg)
else:
print(value)