#
openstack_info = {
'username' : 'xxxxxx',
- 'password' : 'xxxxxx',
+ 'password' : 'xxxxx',
'auth_url' : 'http://10.66.4.102:5000/v2.0/',
- 'project_name' : 'xxxxxx',
+ 'project_name' : 'xxxxx',
'mgmt_network' : 'private',
'reserved_flavor' : 'm1.medium',
'reserved_image' : 'Fedora-x86_64-20-20131211.1-sda-ping.qcow2',
def get_cal_account():
"""
- Creates an object for class RwcalYang.CloudAccount()
+ Creates an object for class RwcalYang.YangData_RwProject_Project_CloudAccounts_CloudAccountList()
"""
- account = RwcalYang.CloudAccount()
+ account = RwcalYang.YangData_RwProject_Project_CloudAccounts_CloudAccountList()
account.name = "Gruntxx"
account.account_type = "openstack"
account.openstack.key = openstack_info['username']
def _get_image_info_request(self):
"""
- Returns request object of type RwcalYang.ImageInfoItem()
+ Returns request object of type RwcalYang.YangData_RwProject_Project_VimResources_ImageinfoList()
"""
- img = RwcalYang.ImageInfoItem()
+ img = RwcalYang.YangData_RwProject_Project_VimResources_ImageinfoList()
img.name = "rift.cal.unittest.image"
img.location = '/net/sharedfiles/home1/common/vm/rift-root-latest.qcow2'
img.disk_format = "qcow2"
def _get_flavor_info_request(self):
"""
- Returns request object of type RwcalYang.FlavorInfoItem()
+ Returns request object of type RwcalYang.YangData_RwProject_Project_VimResources_FlavorinfoList()
"""
- flavor = RwcalYang.FlavorInfoItem()
+ flavor = RwcalYang.YangData_RwProject_Project_VimResources_FlavorinfoList()
flavor.name = 'rift.cal.unittest.flavor'
flavor.vm_flavor.memory_mb = 16384 # 16GB
flavor.vm_flavor.vcpu_count = 4
"""
Returns request object of type RwcalYang.VMInfoItem
"""
- vm = RwcalYang.VMInfoItem()
+ vm = RwcalYang.YangData_RwProject_Project_VimResources_VminfoList()
vm.vm_name = 'rift.cal.unittest.vm'
vm.flavor_id = flavor_id
vm.image_id = image_id
def _get_network_info_request(self):
"""
- Returns request object of type RwcalYang.NetworkInfoItem
+ Returns request object of type RwcalYang.YangData_RwProject_Project_VimResources_NetworkinfoList
"""
- network = RwcalYang.NetworkInfoItem()
+ network = RwcalYang.YangData_RwProject_Project_VimResources_NetworkinfoList()
network.network_name = 'rift.cal.unittest.network'
network.subnet = '192.168.16.0/24'
if openstack_info['physical_network']:
def _get_port_info_request(self, network_id, vm_id):
"""
- Returns an object of type RwcalYang.PortInfoItem
+ Returns an object of type RwcalYang.YangData_RwProject_Project_VimResources_PortinfoList
"""
- port = RwcalYang.PortInfoItem()
+ port = RwcalYang.YangData_RwProject_Project_VimResources_PortinfoList()
port.port_name = 'rift.cal.unittest.port'
port.network_id = network_id
if vm_id != None:
"""
logger.info("Openstack-CAL-Test: Test Get VDU List APIs")
rc, rsp = self.cal.get_vdu_list(self._acct)
- self.assertEqual(rc, RwStatus.SUCCESS)
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
logger.info("Openstack-CAL-Test: Received %d VDUs" %(len(rsp.vdu_info_list)))
for vdu in rsp.vdu_info_list:
- rc, vdu2 = self.cal.get_vdu(self._acct, vdu.vdu_id)
+ rc, vdu2 = self.cal.get_vdu(self._acct, vdu.vdu_id, "")
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
+ # Make changes for the third argument (currently None for mgmt_network).
+ # This is the mgmt_network published in the vdur (vdu.mgmt_network).
+ # Pass accordingly as per the use case of the test.
self.assertEqual(vdu2.vdu_id, vdu.vdu_id)
def _get_virtual_link_request_info(self):
"""
- Returns object of type RwcalYang.VirtualLinkReqParams
+ Returns object of type RwcalYang.YangData_RwProject_Project_VirtualLinkReqParams
"""
- vlink = RwcalYang.VirtualLinkReqParams()
+ vlink = RwcalYang.YangData_RwProject_Project_VirtualLinkReqParams()
vlink.name = 'rift.cal.virtual_link'
vlink.subnet = '192.168.1.0/24'
if openstack_info['physical_network']:
def _get_vdu_request_info(self, virtual_link_id):
"""
- Returns object of type RwcalYang.VDUInitParams
+ Returns object of type RwcalYang.YangData_RwProject_Project_VduInitParams
"""
- vdu = RwcalYang.VDUInitParams()
+ vdu = RwcalYang.YangData_RwProject_Project_VduInitParams()
vdu.name = "cal.vdu"
vdu.node_id = OpenStackTest.NodeID
vdu.image_id = self._image.id
def _get_vdu_modify_request_info(self, vdu_id, virtual_link_id):
"""
- Returns object of type RwcalYang.VDUModifyParams
+ Returns object of type RwcalYang.YangData_RwProject_Project_VduModifyParams
"""
- vdu = RwcalYang.VDUModifyParams()
+ vdu = RwcalYang.YangData_RwProject_Project_VduModifyParams()
vdu.vdu_id = vdu_id
c1 = vdu.connection_points_add.add()
c1.name = "c_modify1"
return vdu
+ def _get_rbsh_vdu_request_info(self, vlink_list):
+ """
+ Returns object of type RwcalYang.YangData_RwProject_Project_VduInitParams
+ """
+ vdu = RwcalYang.YangData_RwProject_Project_VduInitParams()
+ vdu.name = "cal_rbsh_vdu"
+ vdu.vm_flavor.memory_mb = 2048
+ vdu.vm_flavor.vcpu_count = 1
+ vdu.vm_flavor.storage_gb = 10
+ vdu.flavor_id = self._flavor.id
+ vdu.allocate_public_address = True
+ ctr = 0
+ for vl in vlink_list:
+ c1 = vdu.connection_points.add()
+ c1.name = "c_point" + str(ctr)
+ ctr += 1
+ c1.virtual_link_id = vl
+ c1.type_yang = 'VIRTIO'
+
+ vol0 = vdu.volumes.add()
+ vol0.name = "vda"
+ vol0.image = "mgmt.img"
+ vol0.size = 40
+ vol0.boot_priority = 0
+ vol0.device_bus = "virtio"
+ vol0.device_type = "disk"
+
+ vol1 = vdu.volumes.add()
+ vol1.name = "vdb"
+ vol1.image = "segstore.img"
+ vol1.size = 60
+ vol1.boot_priority = 1
+ vol1.device_bus = "virtio"
+ vol1.device_type = "disk"
+
+ # blank disk
+ vol2 = vdu.volumes.add()
+ vol2.name = "vdc"
+ vol2.size = 10
+ vol2.boot_priority = 2
+ vol2.device_bus = "virtio"
+ vol2.device_type = "disk"
+
+ # existing volume disk
+ vol3 = vdu.volumes.add()
+ vol3.name = "vdd"
+ vol3.size = 10
+ vol3.volume_ref = "volume-ref1"
+ vol3.boot_priority = 3
+ vol3.device_bus = "virtio"
+ vol3.device_type = "disk"
+ return vdu
+
+ @unittest.skip("Skipping test_create_rbsh_vdu")
+ def test_create_rbsh_vdu(self):
+ """
+ Test to create VDU with mgmt port and 3 additional connection points
+ """
+ logger.info("Openstack-CAL-Test: Test Create Virtual Link API")
+ vlink_list = []
+ for ctr in range(3):
+ vlink = RwcalYang.YangData_RwProject_Project_VirtualLinkReqParams()
+ vlink.name = 'rift.cal.virtual_link' + str(ctr)
+ vlink.subnet = '11.0.{}.0/24'.format(str(1 + ctr))
+
+ rc, rsp = self.cal.create_virtual_link(self._acct, vlink)
+ self.assertEqual(rc.status, 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)
+ self.assertEqual(rsp.virtual_link_id, vlink_id)
+ vlink_list.append(vlink_id)
+
+
+ # Now create VDU
+ vdu_req = self._get_rbsh_vdu_request_info(vlink_list)
+ logger.info("Openstack-CAL-Test: Test Create RB steelhead VDU API (w/ mgmt port) and 3 CPs")
+
+ rc, rsp = self.cal.create_vdu(self._acct, vdu_req)
+ logger.debug("Openstack-CAL-Test: rc %s rsp %s" % (rc, rsp))
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
+ logger.info("Openstack-CAL-Test: Created vdu with Id: %s" %rsp)
+
+ test_vdu_id = rsp
+
+ ## Check if VDU get is successful
+ rc, rsp = self.cal.get_vdu(self._acct, test_vdu_id, "")
+ # Make changes for the third argument (currently None for mgmt_network).
+ # This is the mgmt_network published in the vdur (vdu.mgmt_network).
+ # Pass accordingly as per the use case of the test.
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
+ logger.debug("Get VDU response %s", rsp)
+ self.assertEqual(rsp.vdu_id, test_vdu_id)
+
+ ### Wait until vdu_state is active
+ logger.debug("Waiting 10 secs")
+ time.sleep(10)
+ #{'name': 'dp0vhost7', 'connection_point_id': 'dp0vhost7', 'state': 'active', 'virtual_link_id': 'rift.cal.virtual_link', 'ip_address': '192.168.100.6'}
+ vdu_state = 'inactive'
+ cp_state = 'inactive'
+ for i in range(15):
+ rc, rsp = self.cal.get_vdu(self._acct, test_vdu_id, "")
+ # Make changes for the third argument (currently None for mgmt_network).
+ # This is the mgmt_network published in the vdur (vdu.mgmt_network).
+ # Pass accordingly as per the use case of the test.
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
+ logger.info("Openstack-CAL-Test: Iter %d VDU with id : %s. Reached State : %s, mgmt ip %s" %(i, test_vdu_id, rsp.state, rsp.management_ip))
+ if (rsp.state == 'active') and ('management_ip' in rsp) and ('public_ip' in rsp):
+ vdu_state = 'active'
+ #'connection_points': [{'name': 'dp0vhost7', 'connection_point_id': 'dp0vhost7', 'state': 'active', 'virtual_link_id': 'rift.cal.virtual_link', 'ip_address': '192.168.100.6'}]
+ for cp in rsp.connection_points:
+ logger.info("Openstack-CAL-Test: Iter %d VDU with id : %s. Reached State : %s CP state %s" %(i, test_vdu_id, rsp.state, cp))
+ logger.debug("Waiting another 5 secs")
+ time.sleep(5)
+
+ self.assertEqual(rc, RwStatus.SUCCESS)
+ self.assertEqual(rsp.state, 'active')
+ self.assertEqual(vdu_state, 'active')
+ logger.info("Openstack-CAL-Test: VDU with id : %s reached expected state : %s IP: %s" %(test_vdu_id, rsp.state, rsp.management_ip))
+ logger.info("Openstack-CAL-Test: VDUInfo: %s" %(rsp))
+ logger.info("Waiting for 30 secs before deletion")
+ time.sleep(30)
+
+ ### Check vdu list as well
+ rc, rsp = self.cal.get_vdu_list(self._acct)
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
+ found = False
+ logger.debug("Get VDU response %s", rsp)
+ for vdu in rsp.vdu_info_list:
+ if vdu.vdu_id == test_vdu_id:
+ found = True
+ self.assertEqual(found, True)
+ logger.info("Openstack-CAL-Test: Passed VDU list" )
+
#@unittest.skip("Skipping test_create_delete_virtual_link_and_vdu")
def test_create_delete_virtual_link_and_vdu(self):
"""
vdu_id = rsp
## Check if VDU create is successful
- rc, rsp = self.cal.get_vdu(self._acct, rsp)
+ rc, rsp = self.cal.get_vdu(self._acct, rsp, "")
+ # Make changes for the third argument (currently None for mgmt_network).
+ # This is the mgmt_network published in the vdur (vdu.mgmt_network).
+ # Pass accordingly as per the use case of the test.
+
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
self.assertEqual(rsp.vdu_id, vdu_id)
### Wait until vdu_state is active
for i in range(50):
- rc, rs = self.cal.get_vdu(self._acct, vdu_id)
- self.assertEqual(rc, RwStatus.SUCCESS)
+ rc, rs = self.cal.get_vdu(self._acct, vdu_id, "")
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
+ # Make changes for the third argument (currently None for mgmt_network).
+ # This is the mgmt_network published in the vdur (vdu.mgmt_network).
+ # Pass accordingly as per the use case of the test.
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
logger.info("Openstack-CAL-Test: VDU with id : %s. Reached State : %s" %(vdu_id, rs.state))
if rs.state == 'active':
break
- rc, rs = self.cal.get_vdu(self._acct, vdu_id)
- self.assertEqual(rc, RwStatus.SUCCESS)
+ rc, rs = self.cal.get_vdu(self._acct, vdu_id, "")
+ # Make changes for the third argument (currently None for mgmt_network).
+ # This is the mgmt_network published in the vdur (vdu.mgmt_network).
+ # Pass accordingly as per the use case of the test.
+ self.assertEqual(rc.status, 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))
vlink_id2= rsp
### Now exercise the modify_vdu_api
- vdu_modify = self._get_vdu_modify_request_info(vdu_id, vlink_id2)
- rc = self.cal.modify_vdu(self._acct, vdu_modify)
- self.assertEqual(rc, RwStatus.SUCCESS)
- logger.info("Openstack-CAL-Test: Modified vdu with Id: %s" %vdu_id)
+ #vdu_modify = self._get_vdu_modify_request_info(vdu_id, vlink_id2)
+ #rc = self.cal.modify_vdu(self._acct, vdu_modify)
+ #self.assertEqual(rc, RwStatus.SUCCESS)
+ #logger.info("Openstack-CAL-Test: Modified vdu with Id: %s" %vdu_id)
### Lets delete the VDU
self.cal.delete_vdu(self._acct, vdu_id)
time.sleep(5)
### Verify that VDU and virtual link are successfully deleted
rc, rsp = self.cal.get_vdu_list(self._acct)
- self.assertEqual(rc, RwStatus.SUCCESS)
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
for vdu in rsp.vdu_info_list:
self.assertNotEqual(vdu.vdu_id, vdu_id)
def _get_vol_vdu_request_info(self, vlink_list):
"""
- Returns object of type RwcalYang.VDUInitParams
+ Returns object of type RwcalYang.YangData_RwProject_Project_VduInitParams
"""
- vdu = RwcalYang.VDUInitParams()
+ vdu = RwcalYang.YangData_RwProject_Project_VduInitParams()
vdu.name = "cal_vdu"
vdu.vm_flavor.memory_mb = 512
vdu.vm_flavor.vcpu_count = 1
"""
logger.info("Openstack-CAL-Test: Test Create Virtual Link API")
vlink_list = []
- vlink = RwcalYang.VirtualLinkReqParams()
+ vlink = RwcalYang.YangData_RwProject_Project_VirtualLinkReqParams()
vlink.name = 'rift.cal.virtual_link'
vlink.subnet = '11.0.1.0/24'
test_vdu_id = rsp
## Check if VDU get is successful
- rc, rsp = self.cal.get_vdu(self._acct, test_vdu_id)
+ rc, rsp = self.cal.get_vdu(self._acct, test_vdu_id, "")
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
+ # Make changes for the third argument (currently None for mgmt_network).
+ # This is the mgmt_network published in the vdur (vdu.mgmt_network).
+ # Pass accordingly as per the use case of the test.
+
logger.debug("Get VDU response %s", rsp)
self.assertEqual(rsp.vdu_id, test_vdu_id)
vdu_state = 'inactive'
cp_state = 'inactive'
for i in range(5):
- rc, rsp = self.cal.get_vdu(self._acct, test_vdu_id)
- self.assertEqual(rc, RwStatus.SUCCESS)
+ rc, rsp = self.cal.get_vdu(self._acct, test_vdu_id, "")
+ # Make changes for the third argument (currently None for mgmt_network).
+ # This is the mgmt_network published in the vdur (vdu.mgmt_network).
+ # Pass accordingly as per the use case of the test.
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
logger.info("Openstack-CAL-Test: VDU with id : %s. Reached State : %s, mgmt ip %s" %(test_vdu_id, rsp.state, rsp.management_ip))
if (rsp.state == 'active') and ('management_ip' in rsp) and ('public_ip' in rsp):
vdu_state = 'active'
### Check vdu list as well
rc, rsp = self.cal.get_vdu_list(self._acct)
- self.assertEqual(rc, RwStatus.SUCCESS)
+ self.assertEqual(rc.status, RwStatus.SUCCESS)
found = False
logger.debug("Get VDU response %s", rsp)
for vdu in rsp.vdu_info_list:
self.assertEqual(found, True)
logger.info("Openstack-CAL-Test: Passed VDU list" )
+ @unittest.skip("Skipping test_validate_creds")
+ def test_validate_creds(self):
+ """
+ Test validate creds
+ """
+ logger.info("Openstack-CAL-Test: Test validate creds")
+ status = self.cal.validate_cloud_creds(self._acct)
+ logger.info("Openstack-CAL-Test: Passed validate creds")
+
class VmData(object):
"""A convenience class that provides all the stats and EPA Attributes
from the VM provided
projects / osm / SO.git / blobdiff