# Copyright 2016 RIFT.IO Inc
#
# Licensed under the Apache License, Version 2.0 (the "License");
# Copyright 2016 RIFT.IO Inc
#
# Licensed under the Apache License, Version 2.0 (the "License");
logger.error("ERROR:Cal plugin instantiation failed. Aborting tests")
else:
logger.info("Openstack Cal plugin successfully instantiated")
logger.error("ERROR:Cal plugin instantiation failed. Aborting tests")
else:
logger.info("Openstack Cal plugin successfully instantiated")
logger.info("Openstack-CAL-Test: setUp")
self.cal = get_cal_plugin()
logger.info("Openstack-CAL-Test: setUpEND")
logger.info("Openstack-CAL-Test: setUp")
self.cal = get_cal_plugin()
logger.info("Openstack-CAL-Test: setUpEND")
# First check for VM Flavor and Image and get the corresponding IDs
rc, rs = self.cal.get_flavor_list(self._acct)
self.assertEqual(rc, RwStatus.SUCCESS)
# First check for VM Flavor and Image and get the corresponding IDs
rc, rs = self.cal.get_flavor_list(self._acct)
self.assertEqual(rc, RwStatus.SUCCESS)
networks = [ network for network in rs.networkinfo_list if (network.network_name == 'rift.cal.unittest.network' or network.network_name == 'rift.cal.virtual_link') ]
for network in networks:
self.cal.delete_virtual_link(self._acct, network.network_id)
networks = [ network for network in rs.networkinfo_list if (network.network_name == 'rift.cal.unittest.network' or network.network_name == 'rift.cal.virtual_link') ]
for network in networks:
self.cal.delete_virtual_link(self._acct, network.network_id)
rc, flv = self.cal.get_flavor(self._acct, flavor.id)
self.assertEqual(rc, RwStatus.SUCCESS)
self.assertEqual(flavor.id, flv.id)
rc, flv = self.cal.get_flavor(self._acct, flavor.id)
self.assertEqual(rc, RwStatus.SUCCESS)
self.assertEqual(flavor.id, flv.id)
# rc, img = self.cal.get_image(self._acct, image.id)
# self.assertEqual(rc, RwStatus.SUCCESS)
# self.assertEqual(image.id, img.id)
# rc, img = self.cal.get_image(self._acct, image.id)
# self.assertEqual(rc, RwStatus.SUCCESS)
# self.assertEqual(image.id, img.id)
for vm in rsp.vminfo_list:
rc, server = self.cal.get_vm(self._acct, vm.vm_id)
self.assertEqual(vm.vm_id, server.vm_id)
for vm in rsp.vminfo_list:
rc, server = self.cal.get_vm(self._acct, vm.vm_id)
self.assertEqual(vm.vm_id, server.vm_id)
for network in rsp.networkinfo_list:
rc, net = self.cal.get_network(self._acct, network.network_id)
self.assertEqual(network.network_id, net.network_id)
for network in rsp.networkinfo_list:
rc, net = self.cal.get_network(self._acct, network.network_id)
self.assertEqual(network.network_id, net.network_id)
flavor = RwcalYang.FlavorInfoItem()
flavor.name = 'rift.cal.unittest.flavor'
flavor.vm_flavor.memory_mb = 16384 # 16GB
flavor = RwcalYang.FlavorInfoItem()
flavor.name = 'rift.cal.unittest.flavor'
flavor.vm_flavor.memory_mb = 16384 # 16GB
flavor.vm_flavor.storage_gb = 40 # 40GB
flavor.guest_epa.mempage_size = OpenStackTest.MemoryPageSize
flavor.guest_epa.cpu_pinning_policy = OpenStackTest.CpuPolicy
flavor.vm_flavor.storage_gb = 40 # 40GB
flavor.guest_epa.mempage_size = OpenStackTest.MemoryPageSize
flavor.guest_epa.cpu_pinning_policy = OpenStackTest.CpuPolicy
if flavor_list:
rc = self.cal.delete_flavor(self._acct, flavor_list[0].id)
self.assertEqual(rc, RwStatus.SUCCESS)
if flavor_list:
rc = self.cal.delete_flavor(self._acct, flavor_list[0].id)
self.assertEqual(rc, RwStatus.SUCCESS)
flavor = self._get_flavor_info_request()
rc, flavor_id = self.cal.create_flavor(self._acct, flavor)
self.assertEqual(rc, RwStatus.SUCCESS)
flavor = self._get_flavor_info_request()
rc, flavor_id = self.cal.create_flavor(self._acct, flavor)
self.assertEqual(rc, RwStatus.SUCCESS)
logger.info("Openstack-CAL-Test: Created new flavor with flavor_id : %s" %(flavor_id))
rc, rs = self.cal.get_flavor(self._acct, flavor_id)
self.assertEqual(rc, RwStatus.SUCCESS)
logger.info("Openstack-CAL-Test: Created new flavor with flavor_id : %s" %(flavor_id))
rc, rs = self.cal.get_flavor(self._acct, flavor_id)
self.assertEqual(rc, RwStatus.SUCCESS)
rc, vm_id = self.cal.create_vm(self._acct, vm)
self.assertEqual(rc, RwStatus.SUCCESS)
rc, vm_id = self.cal.create_vm(self._acct, vm)
self.assertEqual(rc, RwStatus.SUCCESS)
rc, rs = self.cal.get_vm_list(self._acct)
self.assertEqual(rc, RwStatus.SUCCESS)
vm_list = [vm for vm in rs.vminfo_list if vm.vm_id == vm_id]
rc, rs = self.cal.get_vm_list(self._acct)
self.assertEqual(rc, RwStatus.SUCCESS)
vm_list = [vm for vm in rs.vminfo_list if vm.vm_id == vm_id]
self.assertEqual(rc, RwStatus.SUCCESS)
### Ensure that VM state is SHUTOFF
self._check_vm_state(vm_id, 'SHUTOFF')
self.assertEqual(rc, RwStatus.SUCCESS)
### Ensure that VM state is SHUTOFF
self._check_vm_state(vm_id, 'SHUTOFF')
def _start_vm(self, vm_id):
"""
Starts VM and performs validity checks
def _start_vm(self, vm_id):
"""
Starts VM and performs validity checks
### Ensure that VM state is ACTIVE
self._check_vm_state(vm_id, 'ACTIVE')
### Ensure that VM state is ACTIVE
self._check_vm_state(vm_id, 'ACTIVE')
def _reboot_vm(self, vm_id):
"""
Reboot VM and perform validity checks
def _reboot_vm(self, vm_id):
"""
Reboot VM and perform validity checks
logger.info("Openstack-CAL-Test: Starting VM(EPA) create/delete test")
flavor = self._get_flavor_info_request()
logger.info("Openstack-CAL-Test: Starting VM(EPA) create/delete test")
flavor = self._get_flavor_info_request()
rc, flavor_id = self.cal.do_create_flavor(self._acct, flavor)
self.assertEqual(rc, RwStatus.SUCCESS)
flavor.id = flavor_id
rc, flavor_id = self.cal.do_create_flavor(self._acct, flavor)
self.assertEqual(rc, RwStatus.SUCCESS)
flavor.id = flavor_id
logger.info("Openstack-CAL-Test: Deleting a network with id : %s. " %(net_id))
rc = self.cal.delete_network(self._acct, net_id)
self.assertEqual(rc, RwStatus.SUCCESS)
logger.info("Openstack-CAL-Test: Deleting a network with id : %s. " %(net_id))
rc = self.cal.delete_network(self._acct, net_id)
self.assertEqual(rc, RwStatus.SUCCESS)
# Verify that network is no longer available via get_network_list API
rc, rs = self.cal.get_network_list(self._acct)
self.assertEqual(rc, RwStatus.SUCCESS)
network_info = [ network for network in rs.networkinfo_list if network.network_id == net_id ]
self.assertEqual(len(network_info), 0)
logger.info("Openstack-CAL-Test: Successfully deleted Network with id : %s" %(net_id))
# Verify that network is no longer available via get_network_list API
rc, rs = self.cal.get_network_list(self._acct)
self.assertEqual(rc, RwStatus.SUCCESS)
network_info = [ network for network in rs.networkinfo_list if network.network_id == net_id ]
self.assertEqual(len(network_info), 0)
logger.info("Openstack-CAL-Test: Successfully deleted Network with id : %s" %(net_id))
rc, rs = self.cal.get_port_list(self._acct)
self.assertEqual(rc, RwStatus.SUCCESS)
port_list = [ port for port in rs.portinfo_list if port.port_id == port_id ]
rc, rs = self.cal.get_port_list(self._acct)
self.assertEqual(rc, RwStatus.SUCCESS)
port_list = [ port for port in rs.portinfo_list if port.port_id == port_id ]
self.assertEqual(rc, RwStatus.SUCCESS)
self.assertEqual(rs.port_state, expected_state)
logger.info("Openstack-CAL-Test: Port with port_id : %s reached expected state : %s" %(port_id, rs.port_state))
self.assertEqual(rc, RwStatus.SUCCESS)
self.assertEqual(rs.port_state, expected_state)
logger.info("Openstack-CAL-Test: Port with port_id : %s reached expected state : %s" %(port_id, rs.port_state))
def _get_vdu_request_info(self, virtual_link_id):
"""
Returns object of type RwcalYang.VDUInitParams
def _get_vdu_request_info(self, virtual_link_id):
"""
Returns object of type RwcalYang.VDUInitParams
#@unittest.skip("Skipping test_create_delete_virtual_link_and_vdu")
def test_create_delete_virtual_link_and_vdu(self):
"""
#@unittest.skip("Skipping test_create_delete_virtual_link_and_vdu")
def test_create_delete_virtual_link_and_vdu(self):
"""
self.assertEqual(rc, RwStatus.SUCCESS)
logger.info("Openstack-CAL-Test: Created virtual_link with Id: %s" %rsp)
vlink_id = rsp
self.assertEqual(rc, RwStatus.SUCCESS)
logger.info("Openstack-CAL-Test: Created virtual_link with Id: %s" %rsp)
vlink_id = rsp
#Check if virtual_link create is successful
rc, rsp = self.cal.get_virtual_link(self._acct, rsp)
self.assertEqual(rc, RwStatus.SUCCESS)
#Check if virtual_link create is successful
rc, rsp = self.cal.get_virtual_link(self._acct, rsp)
self.assertEqual(rc, RwStatus.SUCCESS)
self.assertEqual(rs.state, 'active')
logger.info("Openstack-CAL-Test: VDU with id : %s reached expected state : %s" %(vdu_id, rs.state))
logger.info("Openstack-CAL-Test: VDUInfo: %s" %(rs))
self.assertEqual(rs.state, 'active')
logger.info("Openstack-CAL-Test: VDU with id : %s reached expected state : %s" %(vdu_id, rs.state))
logger.info("Openstack-CAL-Test: VDUInfo: %s" %(rs))