#docstring
-""" This is the main module for OPNFV Auto Test Data for Use Case 2: Resilience Improvements Through ONAP.
+"""This is the main module for OPNFV Auto Test Data for Use Case 2: Resilience Improvements Through ONAP.
Auto project: https://wiki.opnfv.org/pages/viewpage.action?pageId=12389095
"""
######################################################################
# import statements
+import AutoResilGlobal
from AutoResilMgTestDef import *
# Constants
USE_CASE_NAME = "Resilience Improvements Through ONAP"
+
######################################################################
def show_menu(selected_test_def_ID):
- print("Project ", PROJECT_NAME, ", Use Case: ",USE_CASE_NAME)
+ """Show menu, with a different first line based on current Test Definition selection."""
+
if selected_test_def_ID>0 :
- print("Current test Definition ID: ",selected_test_def_ID)
+ print("\nCurrently selected test Definition ID: ",selected_test_def_ID)
else:
- print("Current test Definition ID: (none)")
+ print("\nCurrently selected test Definition ID: (none)")
print("1: select Test Definition ID")
- print("2: view current Test Definition details")
- print("3: start an execution of current Test Definition")
+ print("2: view currently selected Test Definition details")
+ print("3: start an execution of currently selected Test Definition")
print("4: exit")
def get_menu_choice():
-
+ """Get a user input (a menu entry number)."""
while True:
try:
user_choice = int(input(" Choice: "))
except ValueError:
- print("Invalid choice (must be an integer). Try again.")
+ print(" Invalid choice (must be an integer). Try again.")
continue
if user_choice < 1 or user_choice > 4:
- print("Invalid choice (must be between 1 and 4). Try again.")
+ print(" Invalid choice (must be between 1 and 4). Try again.")
continue
else:
return user_choice
def get_test_def_ID():
-
+ """Get a user input (a test definition ID)."""
while True:
try:
- user_test_def_ID = int(input(" Test Definition ID: "))
+ user_test_def_ID = int(input(" Test Definition ID: "))
except ValueError:
- print("Invalid choice (must be an integer). Try again.")
+ print(" Invalid choice (must be an integer). Try again.")
+ continue
+ if user_test_def_ID <1:
+ print(" Invalid choice (must be a positive integer). Try again.")
continue
test_defs = read_list_bin(FILE_TEST_DEFINITIONS)
if (test_defs == None) or (test_defs==[]):
print("Problem with test definition file: empty")
- sys.exit()
+ sys.exit() # stop entire program, because test definition file MUST be correct
if index_already_there(user_test_def_ID, test_defs):
return user_test_def_ID
else:
- print("Invalid choice (Test Definition ID ",user_test_def_ID," does not exist). Try again.")
+ print("Invalid choice (Test Definition ID",user_test_def_ID,"does not exist). Try again.")
continue
######################################################################
def main():
- # TODO: run initializations to refresh data and make sure files are here
+ print("\nProject:\t", PROJECT_NAME)
+ print("Use Case:\t",USE_CASE_NAME)
+
+ # Run initializations, to refresh data and make sure files are here. Also, this loads the lists in memory.
+ # For now, initialization functions are self-contained and hard-coded:
+ # all definition data is initialized from the code, not from user interaction.
+ AutoResilGlobal.test_case_list = init_test_cases()
+ AutoResilGlobal.test_definition_list = init_test_definitions()
+ AutoResilGlobal.recipient_list = init_recipients()
+ AutoResilGlobal.challenge_definition_list = init_challenge_definitions()
+ AutoResilGlobal.metric_definition_list = init_metric_definitions()
+
+ AutoResilGlobal.physical_resource_list = init_physical_resources()
+ AutoResilGlobal.cloud_virtual_resource_list = init_cloud_virtual_resources()
+ AutoResilGlobal.VNF_Service_list = init_VNFs_Services()
+
+
+ # start with no test definition selected
selected_test_def_ID = -1
+ # interactive menu loop
while True:
show_menu(selected_test_def_ID)
user_choice = get_menu_choice()
- #print("user_choice:",user_choice) #test
+ #print("***user_choice:",user_choice) #debug
- if user_choice == 1:
+ if user_choice == 1: # select Test Definition ID
selected_test_def_ID = get_test_def_ID()
+ selected_test_def = get_indexed_item_from_list(selected_test_def_ID, AutoResilGlobal.test_definition_list)
+ continue
- if user_choice == 4:
+ if user_choice == 2: # view currently selected Test Definition details
+ if selected_test_def_ID > 0:
+ if selected_test_def == None:
+ print("Problem with test definition: empty")
+ sys.exit() # stop entire program, because test definition MUST be correct
+ else:
+ selected_test_def.printout_all(0)
+ continue
+ else:
+ print("No current selection of Test Definition. Try again.")
+ continue
+
+ if user_choice == 3: # start an execution of currently selected Test Definition
+ if selected_test_def_ID > 0:
+ if selected_test_def == None:
+ print("Problem with test definition: empty")
+ sys.exit() # stop entire program, because test definition MUST be correct
+ else:
+ # TODO run test: method of TestDefinition, or function ?
+ pass
+ else:
+ print("No current selection of Test Definition. Try again.")
+ continue
+
+ if user_choice == 4: # exit
+ print("\nEnd of Main Program")
+ print("\nProject:\t", PROJECT_NAME)
+ print("Use Case:\t",USE_CASE_NAME)
+ print("\nBye!\n")
sys.exit()
- print(get_indexed_item_from_file(selected_test_def_ID,FILE_TEST_DEFINTIONS))
- print(get_indexed_item_from_file(5,FILE_TEST_CASES))
- print("End of Main\n Project: \t\t", PROJECT_NAME, "\n Use Case:\t",USE_CASE_NAME)
if __name__ == "__main__":
main()
import sys
from enum import Enum
from datetime import datetime, timedelta
+import AutoResilGlobal
# Constants with definition file names
FILE_PHYSICAL_RESOURCES = "ResourcesPhysical.bin"
FILE_CHALLENGE_DEFINITIONS = "DefinitionsChallenges.bin"
FILE_TEST_DEFINITIONS = "DefinitionsTests.bin"
+# Other constants
+INDENTATION_MULTIPLIER = 4
+
######################################################################
# Auto JIRA link
self.JIRA_URL = test_case_JIRA_URL
+ def printout_all(self, indent_level):
+ """Print out all attributes, with an indentation level."""
+ indent = " "*indent_level*INDENTATION_MULTIPLIER
+
+ print(indent, "Test Case ID:", self.ID, sep='')
+ print(indent, "|-name:", self.name, sep='')
+
+ print(indent, "|-JIRA URL:", self.JIRA_URL, sep='')
+
# no need for functions to remove data: ever-growing library, arbitrary ID
# initial version: should not even add data dynamically, in case object signature changes
self.test_API_command_sent_list = test_def_testAPICommandSent
+ def printout_all(self, indent_level):
+ """Print out all attributes, with an indentation level."""
+ indent = " "*indent_level*INDENTATION_MULTIPLIER
+
+ print(indent, "\nTest Definition ID:", self.ID, sep='')
+ print(indent, "|-name:", self.name, sep='')
+
+ print(indent, "|-associated test case ID:", self.test_case_ID, sep='')
+ test_case = get_indexed_item_from_list(self.test_case_ID, AutoResilGlobal.test_case_list)
+ if test_case != None:
+ test_case.printout_all(indent_level+1)
+
+ print(indent, "|-associated challenge def ID:", self.challenge_def_ID, sep='')
+ challenge_def = get_indexed_item_from_list(self.challenge_def_ID, AutoResilGlobal.challenge_definition_list)
+ if challenge_def != None:
+ challenge_def.printout_all(indent_level+1)
+
+ if self.VNF_ID_list != None:
+ if len(self.VNF_ID_list) >0:
+ print(indent, "|-associated VNFs:", sep='')
+ for VNF_ID in self.VNF_ID_list:
+ VNF_item = get_indexed_item_from_list(VNF_ID, AutoResilGlobal.VNF_Service_list)
+ if VNF_item != None:
+ VNF_item.printout_all(indent_level+1)
+
+ if self.associated_metrics_ID_list != None:
+ if len(self.associated_metrics_ID_list) >0:
+ print(indent, "|-associated metrics:", sep='')
+ for Metric_ID in self.associated_metrics_ID_list:
+ Metric_item = get_indexed_item_from_list(Metric_ID, AutoResilGlobal.metric_definition_list)
+ if Metric_item != None:
+ Metric_item.printout_all(indent_level+1)
+
+ if self.recipient_ID_list != None:
+ if len(self.recipient_ID_list) >0:
+ print(indent, "|-associated recipients:", sep='')
+ for recipient_ID in self.recipient_ID_list:
+ recipient_item = get_indexed_item_from_list(recipient_ID, AutoResilGlobal.recipient_list)
+ if recipient_item != None:
+ recipient_item.printout_all(indent_level+1)
+
+ if self.test_CLI_command_sent_list != None:
+ if len(self.test_CLI_command_sent_list) >0:
+ print(indent, "|-associated CLI commands:", sep='')
+ for CLI_command in self.test_CLI_command_sent_list:
+ print(" "*INDENTATION_MULTIPLIER, "|- ", CLI_command, sep='')
+
+ # TODO: self.test_API_command_sent_list (depends how API commands are stored: likely a list of strings)
+
+
+
def init_test_definitions():
"""Function to initialize test definition data."""
test_definitions = []
test_def_challengeDefID = 1
test_def_testCaseID = 5
test_def_VNFIDs = [1]
- test_def_associatedMetricsIDs = []
+ test_def_associatedMetricsIDs = [2]
test_def_recipientIDs = [2]
- test_def_testCLICommandSent = ["pwd"]
+ test_def_testCLICommandSent = ["pwd","kubectl describe pods --include-uninitialized=false"]
test_def_testAPICommandSent = ["data1","data2"]
test_definitions.append(TestDefinition(test_def_ID, test_def_name,
test_def_challengeDefID,
def __init__ (self, chall_def_ID, chall_def_name,
chall_def_challengeType,
chall_def_recipientID,
- chall_def_impactedResourcesInfo,
- chall_def_impactedResourceIDs,
+ chall_def_impactedCloudResourcesInfo,
+ chall_def_impactedCloudResourceIDs,
+ chall_def_impactedPhysResourcesInfo,
+ chall_def_impactedPhysResourceIDs,
chall_def_startChallengeCLICommandSent,
chall_def_stopChallengeCLICommandSent,
chall_def_startChallengeAPICommandSent,
self.challenge_type = chall_def_challengeType
# recipient instance, to start/stop the challenge
self.recipient_ID = chall_def_recipientID
- # free-form info about impacted resource(s)
- self.impacted_resources_info = chall_def_impactedResourcesInfo
+
+ # free-form info about cloud virtual impacted resource(s)
+ self.impacted_cloud_resources_info = chall_def_impactedCloudResourcesInfo
+ # impacted resources (list of IDs, usually only 1)
+ self.impacted_cloud_resource_ID_list = chall_def_impactedCloudResourceIDs
+
+ # free-form info about physical impacted resource(s)
+ self.impacted_phys_resources_info = chall_def_impactedPhysResourcesInfo
# impacted resources (list of IDs, usually only 1)
- self.impacted_resource_ID_list = chall_def_impactedResourceIDs
+ self.impacted_phys_resource_ID_list = chall_def_impactedPhysResourceIDs
+
# if CLI; can include hard-coded references to resources
self.start_challenge_CLI_command_sent = chall_def_startChallengeCLICommandSent
# if CLI; to restore to normal
self.stop_challenge_API_command_sent = chall_def_stopChallengeAPICommandSent
+ def printout_all(self, indent_level):
+ """Print out all attributes, with an indentation level."""
+ indent = " "*indent_level*INDENTATION_MULTIPLIER
+
+ print(indent, "Challenge Definition ID:", self.ID, sep='')
+ print(indent, "|-name:", self.name, sep='')
+
+ print(indent, "|-challenge type:", self.challenge_type, sep='')
+
+ print(indent, "|-associated recipient ID:", self.recipient_ID, sep='')
+ recipient = get_indexed_item_from_list(self.recipient_ID, AutoResilGlobal.recipient_list)
+ if recipient != None:
+ recipient.printout_all(indent_level+1)
+
+ print(indent, "|-info about cloud virtual impacted resource(s):", self.impacted_cloud_resources_info, sep='')
+
+ if self.impacted_cloud_resource_ID_list != None:
+ if len(self.impacted_cloud_resource_ID_list) >0:
+ print(indent, "|-associated cloud virtual impacted resource(s):", sep='')
+ for cloud_resource_ID in self.impacted_cloud_resource_ID_list:
+ cloud_resource_item = get_indexed_item_from_list(cloud_resource_ID, AutoResilGlobal.cloud_virtual_resource_list)
+ if cloud_resource_item != None:
+ cloud_resource_item.printout_all(indent_level+1)
+
+ print(indent, "|-info about physical virtual impacted resource(s):", self.impacted_phys_resources_info, sep='')
+
+ if self.impacted_phys_resource_ID_list != None:
+ if len(self.impacted_phys_resource_ID_list) >0:
+ print(indent, "|-associated physical impacted resource(s):", sep='')
+ for phys_resource_ID in self.impacted_phys_resource_ID_list:
+ phys_resource_item = get_indexed_item_from_list(phys_resource_ID, AutoResilGlobal.physical_resource_list)
+ if phys_resource_item != None:
+ phys_resource_item.printout_all(indent_level+1)
+
+ print(indent, "|-CLI command to start challenge:", self.start_challenge_CLI_command_sent, sep='')
+
+ print(indent, "|-CLI command to stop challenge:", self.stop_challenge_CLI_command_sent, sep='')
+
+ # TODO: self.start_challenge_API_command_sent (depends how API commands are stored: likely a list of strings)
+ # TODO: self.stop_challenge_API_command_sent (depends how API commands are stored: likely a list of strings)
+
+
+
+
def init_challenge_definitions():
"""Function to initialize challenge definition data."""
challenge_defs = []
chall_def_name = "VM failure"
chall_def_challengeType = ChallengeType.COMPUTE_HOST_FAILURE
chall_def_recipientID = 1
- chall_def_impactedResourcesInfo = "OpenStack VM on ctl02 in Arm pod"
- chall_def_impactedResourceIDs = [2]
+ chall_def_impactedCloudResourcesInfo = "OpenStack VM on ctl02 in Arm pod"
+ chall_def_impactedCloudResourceIDs = [2]
+ chall_def_impactedPhysResourcesInfo = "physical server XYZ"
+ chall_def_impactedPhysResourceIDs = [1]
chall_def_startChallengeCLICommandSent = "service nova-compute stop"
chall_def_stopChallengeCLICommandSent = "service nova-compute restart"
+ # OpenStack VM Suspend vs. Pause: suspend stores the state of VM on disk while pause stores it in memory (RAM)
+ # $ nova suspend NAME
+ # $ nova resume NAME
+
chall_def_startChallengeAPICommandSent = []
chall_def_stopChallengeAPICommandSent = []
challenge_defs.append(ChallengeDefinition(chall_def_ID, chall_def_name,
chall_def_challengeType,
chall_def_recipientID,
- chall_def_impactedResourcesInfo,
- chall_def_impactedResourceIDs,
+ chall_def_impactedCloudResourcesInfo,
+ chall_def_impactedCloudResourceIDs,
+ chall_def_impactedPhysResourcesInfo,
+ chall_def_impactedPhysResourceIDs,
chall_def_startChallengeCLICommandSent,
chall_def_stopChallengeCLICommandSent,
chall_def_startChallengeAPICommandSent,
self.username_creds = recipient_userNameCreds
# optional: password for user/pwd credentials
self.password_creds = recipient_passwordCreds
- # optional: password for user/pwd credentials
+ # optional: key credentials
self.key_creds = recipient_keyCreds
# optional: info about recipient's network (VPN, VCN, VN, Neutron, ...)
self.network_info = recipient_networkInfo
+ def printout_all(self, indent_level):
+ """Print out all attributes, with an indentation level."""
+ indent = " "*indent_level*INDENTATION_MULTIPLIER
+
+ print(indent, "Recipient ID:", self.ID, sep='')
+ print(indent, "|-name:", self.name, sep='')
+
+ print(indent, "|-version info:", self.version_info, sep='')
+ print(indent, "|-IP address:", self.access_IP_address, sep='')
+ print(indent, "|-URL:", self.access_URL, sep='')
+ print(indent, "|-username for user/pwd credentials:", self.username_creds, sep='')
+ print(indent, "|-password for user/pwd credentials:", self.password_creds, sep='')
+ print(indent, "|-key credentials:", self.key_creds, sep='')
+ print(indent, "|-info about network:", self.network_info, sep='')
+
+
+
def init_recipients():
"""Function to initialize recipient data."""
test_recipients = []
recipient_userNameCreds = "ali"
recipient_passwordCreds = "baba"
recipient_keyCreds = "ssh-rsa k7fjsnEFzESfg6phg"
- recipient_networkInfo = "UNH IOL 172.16.0.0/16"
+ recipient_networkInfo = "UNH IOL 172.16.0.0/22"
+
+ test_recipients.append(Recipient(recipient_ID, recipient_name,
+ recipient_info,
+ recipient_versionInfo,
+ recipient_accessIPAddress,
+ recipient_accessURL,
+ recipient_userNameCreds,
+ recipient_passwordCreds,
+ recipient_keyCreds,
+ recipient_networkInfo))
+
+ recipient_ID = 2
+ recipient_name = "Kubernetes on x86 pod"
+ recipient_info = "bare metal"
+ recipient_versionInfo = "v1.9"
+ recipient_accessIPAddress = "8.9.7.6"
+ recipient_accessURL = ""
+ recipient_userNameCreds = "kuber"
+ recipient_passwordCreds = "netes"
+ recipient_keyCreds = "ssh-rsa 0fjs7hjghsa37fhfs"
+ recipient_networkInfo = "UNH IOL 10.10.30.157/22"
+
test_recipients.append(Recipient(recipient_ID, recipient_name,
recipient_info,
self.info = metric_def_info
+ def printout_all(self, indent_level):
+ """Print out all attributes, with an indentation level."""
+ indent = " "*indent_level*INDENTATION_MULTIPLIER
+
+ print(indent, "Metric Definition ID:", self.ID, sep='')
+ print(indent, "|-name:", self.name, sep='')
+
+ print(indent, "|-info:", self.info, sep='')
+
+
class MetricValue:
"""Object for storing a measurement of a Metric Definition for Auto project, with common attributes
(value, timestamp, metric_def_ID).
self.MAC_address = phys_resrc_MACAddress
+ def printout_all(self, indent_level):
+ """Print out all attributes, with an indentation level."""
+ indent = " "*indent_level*INDENTATION_MULTIPLIER
+
+ print(indent, "Physical Resource ID:", self.ID, sep='')
+ print(indent, "|-name:", self.name, sep='')
+
+ print(indent, "|-info:", self.info, sep='')
+ print(indent, "|-IP address:", self.IP_address, sep='')
+ print(indent, "|-MAC address:", self.MAC_address, sep='')
+
+
+
def init_physical_resources():
"""Function to initialize physical resource data."""
test_physical_resources = []
phys_resrc_name = "small-cavium-1"
phys_resrc_info = "Jump server in Arm pod, 48 cores, 64G RAM, 447G SSD, aarch64 Cavium ThunderX, Ubuntu OS"
phys_resrc_IPAddress = "10.10.50.12"
- phys_resrc_MACAddress = ""
+ phys_resrc_MACAddress = "00-14-22-01-23-45"
test_physical_resources.append(PhysicalResource(phys_resrc_ID, phys_resrc_name,
phys_resrc_info,
phys_resrc_IPAddress,
phys_resrc_MACAddress))
+ phys_resrc_ID = 2
+ phys_resrc_name = "medium-cavium-1"
+ phys_resrc_info = "Jump server in New York pod, 96 cores, 64G RAM, 447G SSD, aarch64 Cavium ThunderX, Ubuntu OS"
+ phys_resrc_IPAddress = "30.31.32.33"
+ phys_resrc_MACAddress = "0xb3:22:05:c1:aa:82"
+
+ test_physical_resources.append(PhysicalResource(phys_resrc_ID, phys_resrc_name,
+ phys_resrc_info,
+ phys_resrc_IPAddress,
+ phys_resrc_MACAddress))
+
+ phys_resrc_ID = 3
+ phys_resrc_name = "mega-cavium-666"
+ phys_resrc_info = "Jump server in Las Vegas, 1024 cores, 1024G RAM, 6666G SSD, aarch64 Cavium ThunderX, Ubuntu OS"
+ phys_resrc_IPAddress = "54.53.52.51"
+ phys_resrc_MACAddress = "01-23-45-67-89-ab"
+
+ test_physical_resources.append(PhysicalResource(phys_resrc_ID, phys_resrc_name,
+ phys_resrc_info,
+ phys_resrc_IPAddress,
+ phys_resrc_MACAddress))
+
+
# write list to binary file
write_list_bin(test_physical_resources, FILE_PHYSICAL_RESOURCES)
# optional: related/associated physical resources (if known and useful or interesting, list of integer IDs)
self.related_phys_rsrc_ID_list = cldvirtres_related_phys_rsrcIDs
+ def printout_all(self, indent_level):
+ """Print out all attributes, with an indentation level."""
+ indent = " "*indent_level*INDENTATION_MULTIPLIER
+
+ print(indent, "Cloud Virtual Resource ID:", self.ID, sep='')
+ print(indent, "|-name:", self.name, sep='')
+
+ print(indent, "|-info:", self.info, sep='')
+ print(indent, "|-IP address:", self.IP_address, sep='')
+ print(indent, "|-URL:", self.URL, sep='')
+
+ if self.related_phys_rsrc_ID_list != None:
+ if len(self.related_phys_rsrc_ID_list) >0:
+ print(indent, "|-related/associated physical resource(s):", sep='')
+ for phys_resource_ID in self.related_phys_rsrc_ID_list:
+ phys_resource_item = get_indexed_item_from_list(phys_resource_ID, AutoResilGlobal.physical_resource_list)
+ if phys_resource_item != None:
+ phys_resource_item.printout_all(indent_level+1)
+
def init_cloud_virtual_resources():
"""Function to initialize cloud virtual resource data."""
cldvirtres_URL,
cldvirtres_related_phys_rsrcIDs))
+ cldvirtres_ID = 2
+ cldvirtres_name = "nova-compute-2"
+ cldvirtres_info = "nova VM in LaaS"
+ cldvirtres_IPAddress = "50.60.70.80"
+ cldvirtres_URL = "http://50.60.70.80:8080"
+ cldvirtres_related_phys_rsrcIDs = [2,3]
+
+ test_cldvirt_resources.append(CloudVirtualResource(cldvirtres_ID, cldvirtres_name,
+ cldvirtres_info,
+ cldvirtres_IPAddress,
+ cldvirtres_URL,
+ cldvirtres_related_phys_rsrcIDs))
+
+ cldvirtres_ID = 3
+ cldvirtres_name = "nova-compute-3"
+ cldvirtres_info = "nova VM in x86 pod"
+ cldvirtres_IPAddress = "50.60.70.80"
+ cldvirtres_URL = "http://50.60.70.80:8080"
+ cldvirtres_related_phys_rsrcIDs = [1]
+
+ test_cldvirt_resources.append(CloudVirtualResource(cldvirtres_ID, cldvirtres_name,
+ cldvirtres_info,
+ cldvirtres_IPAddress,
+ cldvirtres_URL,
+ cldvirtres_related_phys_rsrcIDs))
+
+
# write list to binary file
write_list_bin(test_cldvirt_resources, FILE_CLOUD_RESOURCES)
self.related_cloud_virt_rsrc_ID_list = vnf_serv_related_cloudvirt_rsrcIDs
+ def printout_all(self, indent_level):
+ """Print out all attributes, with an indentation level."""
+ indent = " "*indent_level*INDENTATION_MULTIPLIER
+
+ print(indent, "VNF or e2e Service ID:", self.ID, sep='')
+ print(indent, "|-name:", self.name, sep='')
+
+ print(indent, "|-info:", self.info, sep='')
+ print(indent, "|-IP address:", self.IP_address, sep='')
+ print(indent, "|-URL:", self.URL, sep='')
+
+ if self.related_phys_rsrc_ID_list != None:
+ if len(self.related_phys_rsrc_ID_list) >0:
+ print(indent, "|-related/associated physical resource(s):", sep='')
+ for phys_resource_ID in self.related_phys_rsrc_ID_list:
+ phys_resource_item = get_indexed_item_from_list(phys_resource_ID, AutoResilGlobal.physical_resource_list)
+ if phys_resource_item != None:
+ phys_resource_item.printout_all(indent_level+1)
+
+ if self.related_cloud_virt_rsrc_ID_list != None:
+ if len(self.related_cloud_virt_rsrc_ID_list) >0:
+ print(indent, "|-related/associated cloud virtual resource(s):", sep='')
+ for cloud_resource_ID in self.related_cloud_virt_rsrc_ID_list:
+ cloud_resource_item = get_indexed_item_from_list(cloud_resource_ID, AutoResilGlobal.cloud_virtual_resource_list)
+ if cloud_resource_item != None:
+ cloud_resource_item.printout_all(indent_level+1)
+
+
+
def init_VNFs_Services():
"""Function to initialize VNFs and e2e Services data."""
test_VNFs_Services = []
vnf_serv_info = "virtual CPE in Arm pod"
vnf_serv_IPAddress = "5.4.3.2"
vnf_serv_URL = "http://5.4.3.2:8080"
- vnf_serv_related_phys_rsrcIDs = [2,4,6]
- vnf_serv_related_cloudvirt_rsrcIDs = [1,2]
+ vnf_serv_related_phys_rsrcIDs = [1,2]
+ vnf_serv_related_cloudvirt_rsrcIDs = [1]
+
+ test_VNFs_Services.append(VNFService(vnf_serv_ID, vnf_serv_name,
+ vnf_serv_info,
+ vnf_serv_IPAddress,
+ vnf_serv_URL,
+ vnf_serv_related_phys_rsrcIDs,
+ vnf_serv_related_cloudvirt_rsrcIDs))
+
+
+ vnf_serv_ID = 2
+ vnf_serv_name = "vFW-1"
+ vnf_serv_info = "virtual Firewall in x86 pod"
+ vnf_serv_IPAddress = "6.7.8.9"
+ vnf_serv_URL = "http://6.7.8.9:8080"
+ vnf_serv_related_phys_rsrcIDs = [3]
+ vnf_serv_related_cloudvirt_rsrcIDs = [2,3]
test_VNFs_Services.append(VNFService(vnf_serv_ID, vnf_serv_name,
vnf_serv_info,
######################################################################
def dump_all_binaries_to_CSV():
- """Get all content from all binary files, and dump everything in a snapshot CSV file."""
+ """Get all content from all Definition data binary files, and dump everything in a snapshot CSV file."""
## TODO
timenow = datetime.now()
######################################################################
def main():
+
+ # everything here is for unit-testing of this module; not part of actual code
tcs = init_test_cases()
print(tcs)
Code Review / auto.git / commitdiff