import yaml
from git import Repo
-from snaps.openstack.utils import nova_utils
from functest.utils import config
logger = logging.getLogger(__name__)
def __init__(self):
- self.snaps_creds = ""
self.vnf_data_dir = getattr(config.CONF, 'dir_router_data')
self.opnfv_vnf_data_dir = "opnfv-vnf-data/"
self.command_template_dir = "command_template/"
os.remove(self.test_result_json_file)
self.logger.debug("removed %s", self.test_result_json_file)
- def get_nova_client(self):
- nova_client = nova_utils.nova_client(self.snaps_creds)
+ self.cloud = None
- return nova_client
-
- def set_credentials(self, snaps_creds):
- self.snaps_creds = snaps_creds
+ def set_credentials(self, cloud):
+ self.cloud = cloud
def get_address(self, server_name, network_name):
- nova_client = self.get_nova_client()
- servers_list = nova_client.servers.list()
- server = None
-
- for server in servers_list:
- if server.name == server_name:
- break
-
+ server = self.cloud.get_server(server_name)
address = server.addresses[
network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
return address
def get_mac_address(self, server_name, network_name):
- nova_client = self.get_nova_client()
- servers_list = nova_client.servers.list()
- server = None
-
- for server in servers_list:
- if server.name == server_name:
- break
-
+ server = self.cloud.get_server(server_name)
mac_address = server.addresses[network_name][
NOVA_CILENT_NETWORK_INFO_INDEX]["OS-EXT-IPS-MAC:mac_addr"]
return mac_address
- def reboot_vm(self, server_name):
- nova_client = self.get_nova_client()
- servers_list = nova_client.servers.list()
- server = None
-
- for server in servers_list:
- if server.name == server_name:
- break
-
- server.reboot()
-
- return
-
- def delete_vm(self, server_name):
- nova_client = self.get_nova_client()
- servers_list = nova_client.servers.list()
- server = None
-
- for server in servers_list:
- if server.name == server_name:
- nova_client.servers.delete(server)
- break
-
- return
-
def get_blueprint_outputs(self, cfy_manager_ip, deployment_name):
url = "http://%s/deployments/%s/outputs" % (
cfy_manager_ip, deployment_name)
network_list.append(networks[network_name])
return network_list
- def request_vnf_reboot(self, vnf_info_list):
- for vnf in vnf_info_list:
- self.logger.debug("reboot the %s", vnf["vnf_name"])
- self.reboot_vm(vnf["vnf_name"])
-
def request_vm_delete(self, vnf_info_list):
for vnf in vnf_info_list:
self.logger.debug("delete the %s", vnf["vnf_name"])
- self.delete_vm(vnf["vnf_name"])
+ self.cloud.delete_server(vnf["vnf_name"])
def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
target_vnf_name):