You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by sw...@apache.org on 2016/06/13 13:53:42 UTC
[5/7] git commit: updated refs/heads/master to 22c6b47
Nuage VSP : Enhancing Marvin test coverage
Project: http://git-wip-us.apache.org/repos/asf/cloudstack/repo
Commit: http://git-wip-us.apache.org/repos/asf/cloudstack/commit/7531f24d
Tree: http://git-wip-us.apache.org/repos/asf/cloudstack/tree/7531f24d
Diff: http://git-wip-us.apache.org/repos/asf/cloudstack/diff/7531f24d
Branch: refs/heads/master
Commit: 7531f24d7374be18eac04000ffac12ee5c462953
Parents: e972273
Author: Prashanth Manthena <Pr...@alcatel-lucent.com>
Authored: Wed Jun 8 15:35:33 2016 +0200
Committer: Nick Livens <ni...@nuagenetworks.net>
Committed: Fri Jun 10 14:33:09 2016 +0200
----------------------------------------------------------------------
.../plugins/nuagevsp/nuageTestCase.py | 530 +++++++++++--------
.../nuagevsp/test_nuage_password_reset.py | 321 ++++++-----
.../nuagevsp/test_nuage_vpc_internal_lb.py | 471 ++++++++--------
.../plugins/nuagevsp/test_nuage_vpc_network.py | 9 +-
.../plugins/nuagevsp/test_nuage_vsp.py | 10 +-
5 files changed, 704 insertions(+), 637 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/7531f24d/test/integration/plugins/nuagevsp/nuageTestCase.py
----------------------------------------------------------------------
diff --git a/test/integration/plugins/nuagevsp/nuageTestCase.py b/test/integration/plugins/nuagevsp/nuageTestCase.py
index 11032a6..a71945c 100644
--- a/test/integration/plugins/nuagevsp/nuageTestCase.py
+++ b/test/integration/plugins/nuagevsp/nuageTestCase.py
@@ -21,6 +21,7 @@
from marvin.cloudstackTestCase import cloudstackTestCase, unittest
from marvin.lib.base import (EgressFireWallRule,
FireWallRule,
+ Host,
Hypervisor,
Network,
NetworkACL,
@@ -45,6 +46,8 @@ from marvin.cloudstackAPI import restartVPC
import importlib
import logging
import socket
+import sys
+import time
class nuageTestCase(cloudstackTestCase):
@@ -60,6 +63,7 @@ class nuageTestCase(cloudstackTestCase):
cls.api_client = test_client.getApiClient()
cls.db_client = test_client.getDbConnection()
cls.test_data = test_client.getParsedTestDataConfig()
+
# Get Zone, Domain and templates
cls.zone = get_zone(cls.api_client,
zone_name=zone.name if zone else None,
@@ -79,6 +83,9 @@ class nuageTestCase(cloudstackTestCase):
)
cls._cleanup = [cls.service_offering]
+ # Check if the host hypervisor type is simulator
+ cls.isSimulator = Hypervisor.list(cls.api_client, zoneid=cls.zone.id)[0].name == "Simulator"
+
# Get configured Nuage VSP device details
try:
physical_networks = PhysicalNetwork.list(cls.api_client, zoneid=cls.zone.id)
@@ -97,42 +104,44 @@ class nuageTestCase(cloudstackTestCase):
cls.cms_id = cls.nuage_vsp_device.cmsid
except Exception as e:
cls.tearDownClass()
- raise unittest.SkipTest("Warning: Couldn't get configured Nuage VSP device details: %s" % e)
-
- # Check if the host hypervisor type is simulator
- cls.isSimulator = Hypervisor.list(cls.api_client, zoneid=cls.zone.id)[0].name == "Simulator"
+ raise unittest.SkipTest("Warning: Could not get configured Nuage VSP device details - %s" % e)
# VSD is a programmable policy and analytics engine of Nuage VSP SDN platform
# vspk is a Python SDK for Nuage VSP's VSD
- # cms_vspk_wrapper is a library that wraps vspk package
+ # libVSD is a library that wraps vspk package
try:
vspk_module = "vspk." + cls.nuage_vsp_device.apiversion if int(cls.nuage_vsp_device.apiversion[1]) >= 4 \
else "vspk.vsdk." + cls.nuage_vsp_device.apiversion
cls.vsdk = importlib.import_module(vspk_module)
- vspk_utils_module = "vspk.utils" if int(cls.nuage_vsp_device.apiversion[1]) >= 4 \
- else "vspk.vsdk." + cls.nuage_vsp_device.apiversion + ".utils"
- vsdk_utils = importlib.import_module(vspk_utils_module)
- set_log_level = getattr(vsdk_utils, "set_log_level")
- from cms_vspk_wrapper.cms_vspk_wrapper import Cms_vspk_wrapper
- except:
- raise unittest.SkipTest("vspk (and/or) cms_vspk_wrapper import failure")
+ from libVSD import ApiClient, VSDHelpers
+ except Exception as e:
+ cls.tearDownClass()
+ raise unittest.SkipTest("Warning: vspk (and/or) libVSD package import failure - %s" % e)
# Configure VSD session
cls._session = cls.vsdk.NUVSDSession(username=cls.nuage_vsp_device.username,
password=cls.nuage_vsp_device.password,
- enterprise="csp", api_url="https://%s:%d" %
- (cls.nuage_vsp_device.hostname,
+ enterprise="csp",
+ api_url="https://%s:%d" % (cls.nuage_vsp_device.hostname,
cls.nuage_vsp_device.port)
)
cls._session.start()
- # Configure cms_vspk_wrapper session
- cls.log_handler = logging.getLogger("CSLog").handlers[0]
+ # Configure libVSD session
+ root = logging.getLogger()
+ log_handler = logging.StreamHandler(sys.stdout)
+ formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+ log_handler.setFormatter(formatter)
+ root.addHandler(log_handler)
vsd_info = cls.nuage_vsp_device.__dict__
- vsd_info["port"] = str(vsd_info["port"])
- cls.vsd = Cms_vspk_wrapper(vsd_info, cls.log_handler)
-
- set_log_level(logging.INFO)
+ cls.debug("Nuage VSP device (VSD) details - %s" % vsd_info)
+ vsd_api_client = ApiClient(address=vsd_info["hostname"],
+ user=vsd_info["username"],
+ password=vsd_info["password"],
+ version=vsd_info["apiversion"][1] + "." + vsd_info["apiversion"][3]
+ )
+ vsd_api_client.new_session()
+ cls.vsd = VSDHelpers(vsd_api_client)
cls.debug("setUpClass nuageTestCase [DONE]")
@@ -180,17 +189,23 @@ class nuageTestCase(cloudstackTestCase):
return vpc_off
# create_Vpc - Creates VPC with the given VPC offering
- def create_Vpc(self, vpc_offering, cidr="10.1.1.1/16", cleanup=True):
- self.debug("Creating a VPC in the account - %s" % self.account.name)
- self.test_data["vpc"]["name"] = "TestVPC"
- self.test_data["vpc"]["displaytext"] = "TestVPC"
- self.test_data["vpc"]["cidr"] = cidr
+ def create_Vpc(self, vpc_offering, cidr='10.1.0.0/16', testdata=None, account=None, networkDomain=None,
+ cleanup=True):
+ if not account:
+ account = self.account
+ self.debug("Creating a VPC in the account - %s" % account.name)
+ if not testdata:
+ testdata = self.test_data["vpc"]
+ testdata["name"] = "TestVPC-" + cidr + "-" + str(vpc_offering.name)
+ testdata["displaytext"] = "Test VPC"
+ testdata["cidr"] = cidr
vpc = VPC.create(self.api_client,
- self.test_data["vpc"],
+ testdata,
vpcofferingid=vpc_offering.id,
zoneid=self.zone.id,
- account=self.account.name,
- domainid=self.account.domainid
+ account=account.name,
+ domainid=account.domainid,
+ networkDomain=networkDomain
)
self.debug("Created VPC with ID - %s" % vpc.id)
if cleanup:
@@ -223,13 +238,20 @@ class nuageTestCase(cloudstackTestCase):
return nw_off
# create_Network - Creates network with the given Network offering
- def create_Network(self, nw_off, gateway="10.1.1.1", netmask="255.255.255.0", vpc=None, acl_list=None):
- self.debug("Creating a network in the account - %s" % self.account.name)
- self.test_data["network"]["netmask"] = netmask
+ def create_Network(self, nw_off, gateway="10.1.1.1", netmask="255.255.255.0", vpc=None, acl_list=None,
+ testdata=None, account=None, cleanup=True):
+ if not account:
+ account = self.account
+ self.debug("Creating a network in the account - %s" % account.name)
+ if not testdata:
+ testdata = self.test_data["network"]
+ testdata["name"] = "TestNetwork-" + gateway + "-" + str(nw_off.name)
+ testdata["displaytext"] = "Test Network"
+ testdata["netmask"] = netmask
network = Network.create(self.api_client,
- self.test_data["network"],
- accountid=self.account.name,
- domainid=self.account.domainid,
+ testdata,
+ accountid=account.name,
+ domainid=account.domainid,
networkofferingid=nw_off.id,
zoneid=self.zone.id,
gateway=gateway,
@@ -237,7 +259,8 @@ class nuageTestCase(cloudstackTestCase):
aclid=acl_list.id if acl_list else None
)
self.debug("Created network with ID - %s" % network.id)
- self.cleanup.append(network)
+ if cleanup:
+ self.cleanup.append(network)
return network
# upgrade_Network - Upgrades the given network
@@ -259,25 +282,74 @@ class nuageTestCase(cloudstackTestCase):
self.cleanup.remove(network)
self.debug("Deleted Network with ID - %s" % network.id)
- # create_VM - Creates VM in the given network, vm_key - Key for the services on the VM
- def create_VM(self, network, vm_key="virtual_machine", host_id=None, start_vm=True):
- self.debug("Creating VM in network with ID - %s" % network.id)
- self.debug("Passed vm_key - %s" % vm_key)
- self.test_data[vm_key]["zoneid"] = self.zone.id
- self.test_data[vm_key]["template"] = self.template.id
+ # create_VM - Creates VM in the given network(s)
+ def create_VM(self, network_list, host_id=None, start_vm=True, testdata=None, account=None, cleanup=True):
+ network_ids = []
+ if isinstance(network_list, list):
+ for network in network_list:
+ network_ids.append(str(network.id))
+ else:
+ network_ids.append(str(network_list.id))
+ if not account:
+ account = self.account
+ self.debug("Creating VM in network(s) with ID(s) - %s in the account - %s" % (network_ids, account.name))
+ if not testdata:
+ testdata = self.test_data["virtual_machine"]
vm = VirtualMachine.create(self.api_client,
- self.test_data[vm_key],
- accountid=self.account.name,
- domainid=self.account.domainid,
+ testdata,
+ accountid=account.name,
+ domainid=account.domainid,
serviceofferingid=self.service_offering.id,
- networkids=[str(network.id)],
+ templateid=self.template.id,
+ zoneid=self.zone.id,
+ networkids=network_ids,
startvm=start_vm,
hostid=host_id
)
- self.debug("Created VM with ID - %s in network with ID - %s" % (vm.id, network.id))
- self.cleanup.append(vm)
+ self.debug("Created VM with ID - %s in network(s) with ID(s) - %s" % (vm.id, network_ids))
+ if cleanup:
+ self.cleanup.append(vm)
return vm
+ # nic_operation_VM - Performs NIC operations such as add, remove, and update default NIC in the given VM and network
+ def nic_operation_VM(self, vm, network, operation="update"):
+ self.debug("Performing %s NIC operation in VM with ID - %s and network with ID - %s" %
+ (operation, vm.id, network.id))
+ for nic in vm.nic:
+ if nic.networkid == network.id:
+ nic_id = nic.id
+ if operation is "update":
+ vm.update_default_nic(self.api_client, nic_id)
+ self.debug("Updated default NIC to NIC with ID - %s in VM with ID - %s and network with ID - %s" %
+ (nic_id, vm.id, network.id))
+ if operation is "remove":
+ vm.remove_nic(self.api_client, nic_id)
+ self.debug("Removed NIC with ID - %s in VM with ID - %s and network with ID - %s" %
+ (nic_id, vm.id, network.id))
+ if operation is "add":
+ vm.add_nic(self.api_client, network.id)
+ self.debug("Added NIC with ID - %s in VM with ID - %s and network with ID - %s" %
+ (nic_id, vm.id, network.id))
+
+ # migrate_VM - Migrates VM to another host, if available
+ def migrate_VM(self, vm):
+ self.debug("Checking if a host is available for migration...")
+ hosts = Host.listForMigration(self.api_client, virtualmachineid=vm.id)
+ self.assertEqual(isinstance(hosts, list), True,
+ "List hosts should return a valid list"
+ )
+ # Remove the host of current VM from the hosts list
+ hosts[:] = [host for host in hosts if host.id != vm.hostid]
+ if len(hosts) <= 0:
+ self.skipTest("No host available for migration. Test requires at-least 2 hosts")
+ host = hosts[0]
+ self.debug("Migrating VM with ID: %s to Host: %s" % (vm.id, host.id))
+ try:
+ vm.migrate(self.api_client, hostid=host.id)
+ except Exception as e:
+ self.fail("Failed to migrate instance, %s" % e)
+ self.debug("Migrated VM with ID: %s to Host: %s" % (vm.id, host.id))
+
# delete_VM - Deletes the given VM
def delete_VM(self, vm, expunge=True):
self.debug("Deleting VM with ID - %s" % vm.id)
@@ -299,12 +371,14 @@ class nuageTestCase(cloudstackTestCase):
return routers[0]
# acquire_PublicIPAddress - Acquires public IP address for the given network/VPC
- def acquire_PublicIPAddress(self, network, vpc=None):
- self.debug("Associating public IP for network with ID - %s" % network.id)
+ def acquire_PublicIPAddress(self, network, vpc=None, account=None):
+ if not account:
+ account = self.account
+ self.debug("Associating public IP for network with ID - %s in the account - %s" % (network.id, account.name))
public_ip = PublicIPAddress.create(self.api_client,
- accountid=self.account.name,
+ accountid=account.name,
+ domainid=account.domainid,
zoneid=self.zone.id,
- domainid=self.account.domainid,
networkid=network.id if vpc is None else None,
vpcid=vpc.id if vpc else self.vpc.id if hasattr(self, "vpc") else None
)
@@ -312,31 +386,34 @@ class nuageTestCase(cloudstackTestCase):
(public_ip.ipaddress.ipaddress, network.id))
return public_ip
- # create_StaticNatRule_For_VM - Creates static NAT rule on the given public IP for the given network and VM
+ # create_StaticNatRule_For_VM - Creates Static NAT rule on the given public IP for the given VM in the given network
def create_StaticNatRule_For_VM(self, vm, public_ip, network, vmguestip=None):
- self.debug("Enabling static NAT for public IP - %s" % public_ip.ipaddress.ipaddress)
- StaticNATRule.enable(self.api_client,
- ipaddressid=public_ip.ipaddress.id,
- virtualmachineid=vm.id,
- networkid=network.id,
- vmguestip=vmguestip
- )
- self.debug("Static NAT enabled for public IP - %s" % public_ip.ipaddress.ipaddress)
+ self.debug("Enabling Static NAT rule on public IP - %s for VM with ID - %s in network with ID - %s" %
+ (public_ip.ipaddress.ipaddress, vm.id, network.id))
+ static_nat_rule = StaticNATRule.enable(self.api_client,
+ ipaddressid=public_ip.ipaddress.id,
+ virtualmachineid=vm.id,
+ networkid=network.id,
+ vmguestip=vmguestip
+ )
+ self.debug("Static NAT rule enabled on public IP - %s for VM with ID - %s in network with ID - %s" %
+ (public_ip.ipaddress.ipaddress, vm.id, network.id))
+ return static_nat_rule
- # delete_StaticNatRule_For_VM - Deletes static NAT rule on the given public IP for the given VM
- def delete_StaticNatRule_For_VM(self, vm, public_ip):
- self.debug("Disabling static NAT for public IP - %s" % public_ip.ipaddress.ipaddress)
+ # delete_StaticNatRule_For_VM - Deletes Static NAT rule on the given public IP
+ def delete_StaticNatRule_For_VM(self, public_ip):
+ self.debug("Disabling Static NAT rule on public IP - %s" % public_ip.ipaddress.ipaddress)
StaticNATRule.disable(self.api_client,
- ipaddressid=public_ip.ipaddress.id,
- virtualmachineid=vm.id
+ ipaddressid=public_ip.ipaddress.id
)
- self.debug("Static NAT disabled for public IP - %s" % public_ip.ipaddress.ipaddress)
+ self.debug("Static NAT rule disabled on public IP - %s" % public_ip.ipaddress.ipaddress)
- # create_FirewallRule - Creates Ingress firewall rule on the given public IP
+ # create_FirewallRule - Creates (Ingress) Firewall rule on the given Static NAT rule enabled public IP for Isolated
+ # networks
def create_FirewallRule(self, public_ip, rule=None):
if not rule:
rule = self.test_data["ingress_rule"]
- self.debug("Adding an Ingress Firewall rule to make Guest VMs accessible through Static NAT - %s" % rule)
+ self.debug("Adding an (Ingress) Firewall rule to make Guest VMs accessible through Static NAT rule - %s" % rule)
return FireWallRule.create(self.api_client,
ipaddressid=public_ip.ipaddress.id,
protocol=rule["protocol"],
@@ -345,7 +422,7 @@ class nuageTestCase(cloudstackTestCase):
endport=rule["endport"]
)
- # create_EgressFirewallRule - Creates Egress firewall rule on the given public IP
+ # create_EgressFirewallRule - Creates Egress Firewall rule in the given Isolated network
def create_EgressFirewallRule(self, network, rule):
self.debug("Adding an Egress Firewall rule to allow/deny outgoing traffic from Guest VMs - %s" % rule)
return EgressFireWallRule.create(self.api_client,
@@ -366,7 +443,7 @@ class nuageTestCase(cloudstackTestCase):
vpcid=vpc.id
)
- # create_NetworkAclRule - Creates Ingress/Egress network ACL rule in the given network/acl list
+ # create_NetworkAclRule - Creates Ingress/Egress Network ACL rule in the given VPC network/acl list
def create_NetworkAclRule(self, rule, traffic_type="Ingress", network=None, acl_list=None):
self.debug("Adding NetworkACL rule - %s" % rule)
if acl_list:
@@ -383,11 +460,23 @@ class nuageTestCase(cloudstackTestCase):
traffictype=traffic_type
)
- # ssh_into_VM - Gets into the shell of the given VM
- def ssh_into_VM(self, vm, public_ip):
+ # ssh_into_VM - Gets into the shell of the given VM using its public IP
+ def ssh_into_VM(self, vm, public_ip, reconnect=True):
self.debug("SSH into VM with ID - %s on public IP address - %s" % (vm.id, public_ip.ipaddress.ipaddress))
- ssh_client = vm.get_ssh_client(ipaddress=public_ip.ipaddress.ipaddress)
- return ssh_client
+ tries = 0
+ while tries < 3:
+ try:
+ ssh_client = vm.get_ssh_client(ipaddress=public_ip.ipaddress.ipaddress, reconnect=reconnect)
+ except Exception as e:
+ self.debug("Failed to SSH into VM: %s" % e)
+ self.debug("Waiting for the VM to be fully resolved for SSH connection...")
+ time.sleep(120)
+ self.debug("Retrying SSH into VM...")
+ tries += 1
+ continue
+ self.debug("Successful to SSH into VM with ID - %s on public IP address - %s" %
+ (vm.id, public_ip.ipaddress.ipaddress))
+ return ssh_client
# execute_cmd - Executes the given command on the given ssh client
def execute_cmd(self, ssh_client, cmd):
@@ -401,11 +490,12 @@ class nuageTestCase(cloudstackTestCase):
self.debug("SSH client executed command result is None")
return ret_data
- # wget_from_server - Fetches index.html file of web server running with the given public IP
- def wget_from_server(self, public_ip):
+ # wget_from_server - Fetches index.html file from a web server listening on the given public IP address and port
+ def wget_from_server(self, public_ip, port):
import urllib
- self.debug("wget from a http server on public IP address - %s" % public_ip.ipaddress.ipaddress)
- filename, headers = urllib.urlretrieve("http://%s/index.html" % public_ip.ipaddress.ipaddress,
+ self.debug("wget index.html file from a http web server listening on public IP address - %s and port - %s" %
+ (public_ip.ipaddress.ipaddress, port))
+ filename, headers = urllib.urlretrieve("http://%s:%s/index.html" % (public_ip.ipaddress.ipaddress, port),
filename="index.html"
)
return filename, headers
@@ -414,7 +504,7 @@ class nuageTestCase(cloudstackTestCase):
# matches the given provider name and state against the list of providers fetched
def validate_NetworkServiceProvider(self, provider_name, state=None):
"""Validates the Network Service Provider in the Nuage VSP Physical Network"""
- self.debug("Check if the Network Service Provider is created successfully ?")
+ self.debug("Validating the creation and state of Network Service Provider - %s" % provider_name)
providers = NetworkServiceProvider.list(self.api_client,
name=provider_name,
physicalnetworkid=self.vsp_physical_network.id)
@@ -426,15 +516,15 @@ class nuageTestCase(cloudstackTestCase):
)
if state:
self.assertEqual(providers[0].state, state,
- "Network Service Provider state should be in state - %s" % state
+ "Network Service Provider state should be '%s'" % state
)
- self.debug("Network Service Provider creation successfully validated for %s" % provider_name)
+ self.debug("Successfully validated the creation and state of Network Service Provider - %s" % provider_name)
- # validate_VpcOffering - Validates the given VPC offering,
- # matches the given VPC offering name and state against the list of VPC offerings fetched
+ # validate_VpcOffering - Validates the given VPC offering, matches the given VPC offering name and state against the
+ # list of VPC offerings fetched
def validate_VpcOffering(self, vpc_offering, state=None):
"""Validates the VPC offering"""
- self.debug("Check if the VPC offering is created successfully ?")
+ self.debug("Validating the creation and state of VPC offering - %s" % vpc_offering.name)
vpc_offs = VpcOffering.list(self.api_client,
id=vpc_offering.id
)
@@ -446,15 +536,14 @@ class nuageTestCase(cloudstackTestCase):
)
if state:
self.assertEqual(vpc_offs[0].state, state,
- "VPC offering state should be in state - %s" % state
+ "VPC offering state should be '%s'" % state
)
- self.debug("VPC offering creation successfully validated for %s" % vpc_offering.name)
+ self.debug("Successfully validated the creation and state of VPC offering - %s" % vpc_offering.name)
- # validate_Vpc - Validates the given VPC,
- # matches the given VPC name and state against the list of VPCs fetched
+ # validate_Vpc - Validates the given VPC, matches the given VPC name and state against the list of VPCs fetched
def validate_Vpc(self, vpc, state=None):
"""Validates the VPC"""
- self.debug("Check if the VPC is created successfully ?")
+ self.debug("Validating the creation and state of VPC - %s" % vpc.name)
vpcs = VPC.list(self.api_client,
id=vpc.id
)
@@ -466,15 +555,15 @@ class nuageTestCase(cloudstackTestCase):
)
if state:
self.assertEqual(vpcs[0].state, state,
- "VPC state should be in state - %s" % state
+ "VPC state should be '%s'" % state
)
- self.debug("VPC creation successfully validated for %s" % vpc.name)
+ self.debug("Successfully validated the creation and state of VPC - %s" % vpc.name)
- # validate_NetworkOffering - Validates the given Network offering,
- # matches the given network offering name and state against the list of network offerings fetched
+ # validate_NetworkOffering - Validates the given Network offering, matches the given network offering name and state
+ # against the list of network offerings fetched
def validate_NetworkOffering(self, net_offering, state=None):
"""Validates the Network offering"""
- self.debug("Check if the Network offering is created successfully ?")
+ self.debug("Validating the creation and state of Network offering - %s" % net_offering.name)
net_offs = NetworkOffering.list(self.api_client,
id=net_offering.id
)
@@ -486,15 +575,15 @@ class nuageTestCase(cloudstackTestCase):
)
if state:
self.assertEqual(net_offs[0].state, state,
- "Network offering state should be in state - %s" % state
+ "Network offering state should be '%s'" % state
)
- self.debug("Network offering creation successfully validated for %s" % net_offering.name)
+ self.debug("Successfully validated the creation and state of Network offering - %s" % net_offering.name)
- # validate_Network - Validates the given network,
- # matches the given network name and state against the list of networks fetched
+ # validate_Network - Validates the given network, matches the given network name and state against the list of
+ # networks fetched
def validate_Network(self, network, state=None):
- """Validates the Network"""
- self.debug("Check if the network is created successfully ?")
+ """Validates the network"""
+ self.debug("Validating the creation and state of Network - %s" % network.name)
networks = Network.list(self.api_client,
id=network.id
)
@@ -506,14 +595,14 @@ class nuageTestCase(cloudstackTestCase):
)
if state:
self.assertEqual(networks[0].state, state,
- "Network state should be in state - %s" % state
+ "Network state should be '%s'" % state
)
- self.debug("Network creation successfully validated for %s" % network.name)
+ self.debug("Successfully validated the creation and state of Network - %s" % network.name)
# check_VM_state - Checks if the given VM is in the expected state form the list of fetched VMs
def check_VM_state(self, vm, state=None):
"""Validates the VM state"""
- self.debug("Check if the VM instance is in state - %s" % state)
+ self.debug("Validating the deployment and state of VM - %s" % vm.name)
vms = VirtualMachine.list(self.api_client,
id=vm.id,
listall=True
@@ -525,12 +614,12 @@ class nuageTestCase(cloudstackTestCase):
self.assertEqual(vms[0].state, state,
"Virtual machine is not in the expected state"
)
- self.debug("Virtual machine instance - %s is in the expected state - %s" % (vm.name, state))
+ self.debug("Successfully validated the deployment and state of VM - %s" % vm.name)
# check_Router_state - Checks if the given router is in the expected state form the list of fetched routers
def check_Router_state(self, router, state=None):
"""Validates the Router state"""
- self.debug("Check if the virtual router instance is in state - %s" % state)
+ self.debug("Validating the deployment and state of Router - %s" % router.name)
routers = Router.list(self.api_client,
id=router.id,
listall=True
@@ -542,13 +631,13 @@ class nuageTestCase(cloudstackTestCase):
self.assertEqual(routers[0].state, state,
"Virtual router is not in the expected state"
)
- self.debug("Virtual router instance - %s is in the expected state - %s" % (router.name, state))
+ self.debug("Successfully validated the deployment and state of Router - %s" % router.name)
# validate_PublicIPAddress - Validates if the given public IP address is in the expected state form the list of
# fetched public IP addresses
def validate_PublicIPAddress(self, public_ip, network, static_nat=False, vm=None):
"""Validates the Public IP Address"""
- self.debug("Check if the public IP is successfully assigned to the network ?")
+ self.debug("Validating the assignment and state of public IP address - %s" % public_ip.ipaddress.ipaddress)
public_ips = PublicIPAddress.list(self.api_client,
id=public_ip.ipaddress.id,
networkid=network.id,
@@ -566,20 +655,22 @@ class nuageTestCase(cloudstackTestCase):
)
if static_nat and vm:
self.assertEqual(public_ips[0].virtualmachineid, vm.id,
- "Static NAT Rule not enabled for the VM using the assigned public IP"
+ "Static NAT rule is not enabled for the VM on the assigned public IP"
)
- self.debug("Assigned Public IP address - %s is successfully validated" % public_ip.ipaddress.ipaddress)
+ self.debug("Successfully validated the assignment and state of public IP address - %s" %
+ public_ip.ipaddress.ipaddress)
# VSD verifications
# VSD is a programmable policy and analytics engine of Nuage VSP SDN platform
- # get_externalID - Returns corresponding external ID of the given object in VSD
- def get_externalID(self, object_id):
- return object_id + "@" + self.cms_id
+ # get_externalID_filter - Returns corresponding external ID filter of the given object in VSD
+ def get_externalID_filter(self, object_id):
+ ext_id = object_id + "@" + self.cms_id
+ return self.vsd.set_externalID_filter(ext_id)
# fetch_by_externalID - Returns VSD object with the given external ID
def fetch_by_externalID(self, fetcher, *cs_objects):
- """ Fetches a child object by external ID using the given fetcher, and uuids of the given cloudstack objects.
+ """ Fetches a child object by external id using the given fetcher, and uuids of the given cloudstack objects.
E.G.
- fetch_by_external_id(vsdk.NUSubnet(id="954de425-b860-410b-be09-c560e7dbb474").vms, cs_vm)
- fetch_by_external_id(session.user.floating_ips, cs_network, cs_public_ip)
@@ -589,162 +680,148 @@ class nuageTestCase(cloudstackTestCase):
"""
return fetcher.get_first(filter="externalID BEGINSWITH '%s'" % ":".join([o.id for o in cs_objects]))
- # verify_vsp_network - Verifies the given domain and network/VPC
- # against the corresponding installed enterprise, domain, zone, and subnet in VSD
- def verify_vsp_network(self, domain_id, network, vpc=None):
- vsd_enterprise = self.vsd.get_enterprise(name=domain_id)
- if vpc:
- ext_network_id = self.get_externalID(vpc.id)
- else:
- ext_network_id = self.get_externalID(network.id)
- ext_subnet_id = self.get_externalID(network.id)
- vsd_domain = self.vsd.get_domain(externalID=ext_network_id)
- vsd_zone = self.vsd.get_zone(externalID=ext_network_id)
- vsd_subnet = self.vsd.get_subnet(externalID=ext_subnet_id)
- self.debug("SHOW ENTERPRISE DATA FORMAT IN VSD")
- self.debug(vsd_enterprise)
- self.assertNotEqual(vsd_enterprise, None,
- "VSD Enterprise data format should not be a None type"
- )
- self.debug("SHOW NETWORK DATA FORMAT IN VSD")
- self.debug(vsd_domain)
- self.debug(vsd_zone)
- self.debug(vsd_subnet)
+ # verify_vsd_network - Verifies the given domain and network/VPC against the corresponding installed enterprise,
+ # domain, zone, and subnet in VSD
+ def verify_vsd_network(self, domain_id, network, vpc=None):
+ self.debug("Verifying the creation and state of Network - %s in VSD" % network.name)
+ vsd_enterprise = self.vsd.get_enterprise(filter=self.get_externalID_filter(domain_id))
+ ext_network_filter = self.get_externalID_filter(vpc.id) if vpc else self.get_externalID_filter(network.id)
+ vsd_domain = self.vsd.get_domain(filter=ext_network_filter)
+ vsd_zone = self.vsd.get_zone(filter=ext_network_filter)
+ vsd_subnet = self.vsd.get_subnet(filter=self.get_externalID_filter(network.id))
+ self.assertEqual(vsd_enterprise.name, domain_id,
+ "VSD enterprise name should match CloudStack domain uuid"
+ )
if vpc:
- self.assertEqual(vsd_domain["description"], "VPC_" + vpc.name,
+ self.assertEqual(vsd_domain.description, "VPC_" + vpc.name,
"VSD domain description should match VPC name in CloudStack"
)
- self.assertEqual(vsd_zone["description"], "VPC_" + vpc.name,
+ self.assertEqual(vsd_zone.description, "VPC_" + vpc.name,
"VSD zone description should match VPC name in CloudStack"
)
else:
- self.assertEqual(vsd_domain["description"], network.name,
+ self.assertEqual(vsd_domain.description, network.name,
"VSD domain description should match network name in CloudStack"
)
- self.assertEqual(vsd_zone["description"], network.name,
+ self.assertEqual(vsd_zone.description, network.name,
"VSD zone description should match network name in CloudStack"
)
- self.assertEqual(vsd_subnet["description"], network.name,
+ self.assertEqual(vsd_subnet.description, network.name,
"VSD subnet description should match network name in CloudStack"
)
-
- # verify_vsp_vm - Verifies the given VM deployment and state in VSD
- def verify_vsp_vm(self, vm, stopped=None):
- ext_vm_id = self.get_externalID(vm.id)
+ self.debug("Successfully verified the creation and state of Network - %s in VSD" % network.name)
+
+ # verify_vsd_vm - Verifies the given VM deployment and state in VSD
+ def verify_vsd_vm(self, vm, stopped=None):
+ self.debug("Verifying the deployment and state of VM - %s in VSD" % vm.name)
+ vsd_vm = self.vsd.get_vm(filter=self.get_externalID_filter(vm.id))
+ self.assertNotEqual(vsd_vm, None,
+ "VM data format in VSD should not be of type None"
+ )
for nic in vm.nic:
- ext_network_id = self.get_externalID(nic.networkid)
- ext_nic_id = self.get_externalID(nic.id)
- vsd_vport = self.vsd.get_vport(subnet_externalID=ext_network_id, vport_externalID=ext_nic_id)
- vsd_vm_interface = self.vsd.get_vm_interface(externalID=ext_nic_id)
- self.debug("SHOW VPORT and VM INTERFACE DATA FORMAT IN VSD")
- self.debug(vsd_vport)
- self.debug(vsd_vm_interface)
- self.assertEqual(vsd_vport["active"], True,
+ vsd_subnet = self.vsd.get_subnet(filter=self.get_externalID_filter(nic.networkid))
+ vsd_vport = self.vsd.get_vport(subnet=vsd_subnet, filter=self.get_externalID_filter(nic.id))
+ vsd_vm_interface = self.vsd.get_vm_interface(filter=self.get_externalID_filter(nic.id))
+ self.assertEqual(vsd_vport.active, True,
"VSD VM vport should be active"
)
- self.assertEqual(vsd_vm_interface["IPAddress"], nic.ipaddress,
+ self.assertEqual(vsd_vm_interface.ip_address, nic.ipaddress,
"VSD VM interface IP address should match VM's NIC IP address in CloudStack"
)
- vsd_vm = self.vsd.get_vm(externalID=ext_vm_id)
- self.debug("SHOW VM DATA FORMAT IN VSD")
- self.debug(vsd_vm)
if not self.isSimulator:
if stopped:
- self.assertEqual(vsd_vm["status"], "DELETE_PENDING",
+ self.assertEqual(vsd_vm.status, "DELETE_PENDING",
"VM state in VSD should be DELETE_PENDING"
)
else:
- self.assertEqual(vsd_vm["status"], vm.state.upper(),
+ self.assertEqual(vsd_vm.status, vm.state.upper(),
"VM state in VSD should match its state in CloudStack"
)
-
- # verify_vsp_router - Verifies the given network router deployment and state in VSD
- def verify_vsp_router(self, router, stopped=None):
- ext_router_id = self.get_externalID(router.id)
- vsd_router = self.vsd.get_vm(externalID=ext_router_id)
- self.debug("SHOW VIRTUAL ROUTER DATA FORMAT IN VSD")
- self.debug(vsd_router)
+ self.debug("Successfully verified the deployment and state of VM - %s in VSD" % vm.name)
+
+ # verify_vsd_router - Verifies the given network router deployment and state in VSD
+ def verify_vsd_router(self, router, stopped=None):
+ self.debug("Verifying the deployment and state of Router - %s in VSD" % router.name)
+ vsd_router = self.vsd.get_vm(filter=self.get_externalID_filter(router.id))
+ self.assertNotEqual(vsd_router, None,
+ "Router data format in VSD should not be of type None"
+ )
if not self.isSimulator:
if stopped:
- self.assertEqual(vsd_router["status"], "DELETE_PENDING",
+ self.assertEqual(vsd_router.status, "DELETE_PENDING",
"Router state in VSD should be DELETE_PENDING"
)
else:
- self.assertEqual(vsd_router["status"], router.state.upper(),
+ self.assertEqual(vsd_router.status, router.state.upper(),
"Router state in VSD should match its state in CloudStack"
)
-
- # verify_vsp_LB_device - Verifies the given LB device deployment and state in VSD
- def verify_vsp_LB_device(self, lb_device, stopped=None):
- ext_lb_device_id = self.get_externalID(lb_device.id)
- vsd_lb_device = self.vsd.get_vm(externalID=ext_lb_device_id)
- self.debug("SHOW LB Device DATA FORMAT IN VSD")
- self.debug(vsd_lb_device)
+ self.debug("Successfully verified the deployment and state of Router - %s in VSD" % router.name)
+
+ # verify_vsd_lb_device - Verifies the given LB device deployment and state in VSD
+ def verify_vsd_lb_device(self, lb_device, stopped=None):
+ self.debug("Verifying the deployment and state of LB device - %s in VSD" % lb_device.name)
+ vsd_lb_device = self.vsd.get_vm(filter=self.get_externalID_filter(lb_device.id))
+ self.assertNotEqual(vsd_lb_device, None,
+ "LB device data format in VSD should not be of type None"
+ )
if not self.isSimulator:
if stopped:
- self.assertEqual(vsd_lb_device['status'], "DELETE_PENDING",
+ self.assertEqual(vsd_lb_device.status, "DELETE_PENDING",
"LB device state in VSD should be DELETE_PENDING"
)
else:
- self.assertEqual(vsd_lb_device['status'], lb_device.state.upper(),
+ self.assertEqual(vsd_lb_device.status, lb_device.state.upper(),
"LB device state in VSD should match its state in CloudStack"
)
-
- # verify_vsp_floating_ip - Verifies the static nat rule on the given public IP of the given network and VM
- # against the corresponding installed FIP in VSD
- def verify_vsp_floating_ip(self, network, vm, public_ipaddress, vpc=None):
- if vpc:
- ext_fip_id = self.get_externalID(vpc.id + ":" + public_ipaddress.id)
- else:
- ext_fip_id = self.get_externalID(network.id + ":" + public_ipaddress.id)
- vsd_fip = self.vsd.get_floating_ip(externalID=ext_fip_id)
- self.debug("SHOW FLOATING IP DATA FORMAT IN VSD")
- self.debug(vsd_fip)
- self.assertEqual(vsd_fip["address"], public_ipaddress.ipaddress,
+ self.debug("Successfully verified the deployment and state of LB device - %s in VSD" % lb_device.name)
+
+ # verify_vsd_floating_ip - Verifies the Static NAT rule on the given public IP of the given VM in the given network
+ # against the corresponding installed Floating IP in VSD
+ def verify_vsd_floating_ip(self, network, vm, public_ipaddress, vpc=None):
+ self.debug("Verifying the assignment and state of public IP address - %s in VSD" % public_ipaddress.ipaddress)
+ ext_fip_filter = self.get_externalID_filter(vpc.id + ":" + public_ipaddress.id) if vpc else \
+ self.get_externalID_filter(network.id + ":" + public_ipaddress.id)
+ vsd_fip = self.vsd.get_floating_ip(filter=ext_fip_filter)
+ self.assertEqual(vsd_fip.address, public_ipaddress.ipaddress,
"Floating IP address in VSD should match acquired public IP address in CloudStack"
)
- if vpc:
- ext_network_id = self.get_externalID(vpc.id)
- else:
- ext_network_id = self.get_externalID(network.id)
- vsd_domain = self.vsd.get_domain(externalID=ext_network_id)
- self.debug("SHOW NETWORK DATA FORMAT IN VSD")
- self.debug(vsd_domain)
- self.assertEqual(vsd_domain["ID"], vsd_fip["parentID"],
- "Floating IP in VSD should be associated with the correct VSD domain, "
- "which in turn should correspond to the correct VPC (or) network in CloudStack"
+ self.assertEqual(vsd_fip.assigned, True,
+ "Floating IP in VSD should be assigned"
+ )
+ ext_network_filter = self.get_externalID_filter(vpc.id) if vpc else self.get_externalID_filter(network.id)
+ vsd_domain = self.vsd.get_domain(filter=ext_network_filter)
+ self.assertEqual(vsd_domain.id, vsd_fip.parent_id,
+ "Floating IP in VSD should be associated with the correct VSD domain, which in turn should "
+ "correspond to the correct VPC (or) network in CloudStack"
)
- ext_subnet_id = self.get_externalID(network.id)
- vsd_subnet = self.vsd.get_subnet(externalID=ext_subnet_id)
+ vsd_subnet = self.vsd.get_subnet(filter=self.get_externalID_filter(network.id))
for nic in vm.nic:
- if nic.networkname == vsd_subnet["description"]:
- ext_network_id = self.get_externalID(nic.networkid)
- ext_nic_id = self.get_externalID(nic.id)
- vsd_vport = self.vsd.get_vport(subnet_externalID=ext_network_id, vport_externalID=ext_nic_id)
- self.debug("SHOW VM VPORT DATA FORMAT IN VSD")
- self.debug(vsd_vport)
- self.assertEqual(vsd_vport["associatedFloatingIPID"], vsd_fip["ID"],
- "Floating IP in VSD should be associated to the correct VSD vport, "
- "which in turn should correspond to the correct Static NAT enabled VM "
- "and network in CloudStack"
+ if nic.networkid == network.id:
+ vsd_vport = self.vsd.get_vport(subnet=vsd_subnet, filter=self.get_externalID_filter(nic.id))
+ self.assertEqual(vsd_vport.associated_floating_ip_id, vsd_fip.id,
+ "Floating IP in VSD should be associated to the correct VSD vport, which in turn should "
+ "correspond to the correct Static NAT rule enabled VM and network in CloudStack"
+ )
+ self.debug("Successfully verified the assignment and state of public IP address - %s in VSD" %
+ public_ipaddress.ipaddress)
+
+ # verify_vsd_firewall_rule - Verifies the given Network Firewall (Ingress/Egress ACL) rule against the corresponding
+ # installed firewall rule in VSD
+ def verify_vsd_firewall_rule(self, firewall_rule, traffic_type="Ingress"):
+ self.debug("Verifying the creation and state of Network Firewall (Ingress/Egress ACL) rule with ID - %s in VSD"
+ % firewall_rule.id)
+ ext_fw_rule_filter = self.get_externalID_filter(firewall_rule.id)
+ vsd_fw_rule = self.vsd.get_egress_acl_entry(filter=ext_fw_rule_filter) if traffic_type is "Ingress" else \
+ self.vsd.get_ingress_acl_entry(filter=ext_fw_rule_filter)
+ self.assertEqual(vsd_fw_rule.policy_state, "LIVE",
+ "Ingress/Egress ACL rule's policy state in VSD should be LIVE"
)
-
- # verify_vsp_firewall_rule - Verifies the given Ingress/Egress firewall rule
- # against the corresponding installed firewall rule in VSD
- def verify_vsp_firewall_rule(self, firewall_rule, traffic_type="Ingress"):
- ext_fw_id = self.get_externalID(firewall_rule.id)
- if traffic_type is "Ingress":
- vsd_fw_rule = self.vsd.get_egress_acl_entry(externalID=ext_fw_id)
- else:
- vsd_fw_rule = self.vsd.get_ingress_acl_entry(externalID=ext_fw_id)
- self.debug("SHOW ACL ENTRY IN VSD")
- self.debug(vsd_fw_rule)
dest_port = str(firewall_rule.startport) + "-" + str(firewall_rule.endport)
- self.assertEqual(vsd_fw_rule["destinationPort"], dest_port,
- "Destination port in VSD should match destination port in CloudStack"
+ self.assertEqual(vsd_fw_rule.destination_port, dest_port,
+ "Ingress/Egress ACL rule's destination port in VSD should match corresponding rule's "
+ "destination port in CloudStack"
)
- vsd_protocol = str(vsd_fw_rule["protocol"])
- self.debug("vsd protocol - %s" % vsd_protocol)
+ vsd_protocol = int(vsd_fw_rule.protocol)
protocol = "tcp"
if vsd_protocol == 6:
protocol = "tcp"
@@ -753,5 +830,8 @@ class nuageTestCase(cloudstackTestCase):
elif vsd_protocol == 17:
protocol = "udp"
self.assertEqual(protocol, firewall_rule.protocol.lower(),
- "Protocol in VSD should match protocol in CloudStack"
+ "Ingress/Egress ACL rule's protocol in VSD should match corresponding rule's protocol in "
+ "CloudStack"
)
+ self.debug("Successfully verified the creation and state of Network Firewall (Ingress/Egress ACL) rule with ID "
+ "- %s in VSD" % firewall_rule.id)
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/7531f24d/test/integration/plugins/nuagevsp/test_nuage_password_reset.py
----------------------------------------------------------------------
diff --git a/test/integration/plugins/nuagevsp/test_nuage_password_reset.py b/test/integration/plugins/nuagevsp/test_nuage_password_reset.py
index ee028c2..bd8bba6 100644
--- a/test/integration/plugins/nuagevsp/test_nuage_password_reset.py
+++ b/test/integration/plugins/nuagevsp/test_nuage_password_reset.py
@@ -24,7 +24,6 @@ from marvin.lib.base import (Account,
VirtualMachine,
Volume)
from marvin.lib.common import list_templates
-from marvin.lib.utils import cleanup_resources
from marvin.cloudstackAPI import updateTemplate
# Import System Modules
from nose.plugins.attrib import attr
@@ -41,130 +40,100 @@ class TestNuagePasswordReset(nuageTestCase):
return
def setUp(self):
- self.cleanup = []
- self.apiclient = self.testClient.getApiClient()
-
- self.account = Account.create(
- self.apiclient,
- self.test_data["account"],
- admin=True,
- domainid=self.domain.id
- )
-
- self.cleanup.append(self.account)
- self.remove_vm2 = False
+ # Create an account
+ self.account = Account.create(self.api_client,
+ self.test_data["account"],
+ admin=True,
+ domainid=self.domain.id
+ )
+ self.cleanup = [self.account]
return
- # tearDown() - Cleans up the setup, removes the VMs
- def tearDown(self):
- self.debug("CLEANUP: TEARDOWN")
- self.apiclient = self.testClient.getApiClient()
- self.updateTemplate(self.defaultTemplateVal)
- self.vm_1.delete(self.apiclient, expunge=True)
- if self.remove_vm2:
- self.vm_2.delete(self.apiclient, expunge=True)
- try:
- cleanup_resources(self.apiclient, self.cleanup)
- except Exception as e:
- self.debug("Warning: Exception during cleanup: %s" % e)
- return
-
- # create_template - Creates template with the given VM object
+ # create_template - Creates guest VM template with the given VM object
def create_template(self, vm):
- self.debug("Creating template")
- list_volume = Volume.list(self.apiclient,
+ self.debug("Creating guest VM template")
+ list_volume = Volume.list(self.api_client,
virtualmachineid=vm.id,
type='ROOT',
- listall=True)
+ listall=True
+ )
if isinstance(list_volume, list):
self.volume = list_volume[0]
else:
raise Exception("Exception: Unable to find root volume for VM with ID - %s" % vm.id)
- self.pw_enabled_template = Template.create(
- self.apiclient,
- self.test_data["template"],
- self.volume.id,
- account=self.account.name,
- domainid=self.account.domainid
- )
- self.assertEqual(self.pw_enabled_template.passwordenabled, True, "template is not passwordenabled")
+ self.pw_enabled_template = Template.create(self.api_client,
+ self.test_data["template"],
+ self.volume.id,
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
+ self.assertEqual(self.pw_enabled_template.passwordenabled, True,
+ "template is not passwordenabled"
+ )
self.cleanup.append(self.pw_enabled_template)
- self.debug("Created template")
+ self.debug("Created guest VM template")
- # updateTemplate - Updates value of template's password enabled setting
+ # updateTemplate - Updates value of the guest VM template's password enabled setting
def updateTemplate(self, value):
- self.debug("Updating value of template's password enabled setting")
+ self.debug("Updating value of guest VM template's password enabled setting")
cmd = updateTemplate.updateTemplateCmd()
cmd.id = self.template.id
cmd.passwordenabled = value
- self.apiclient.updateTemplate(cmd)
- list_template_response = list_templates(self.apiclient,
+ self.api_client.updateTemplate(cmd)
+ list_template_response = list_templates(self.api_client,
templatefilter="all",
id=self.template.id
)
self.template = list_template_response[0]
- self.debug("Updated template")
-
- # VM object is passed as an argument and its interface id is returned
- def get_vm_interface_id(self, vm):
- self.debug("GET VM INTERFACE ID")
- nic_ext_id = self.get_externalID(vm.nic[0].id)
- vm_interface = self.vsd.get_vm_interface(externalID=nic_ext_id)
- vm_interface_id = vm_interface["ID"]
- return vm_interface_id
+ self.debug("Updated guest VM template")
- # VM object is passed as an argument and its userdata URL is returned
+ # get_userdata_url - Returns user data URL for the given VM object
def get_userdata_url(self, vm):
- self.debug("GET USER DATA URL")
+ self.debug("Getting user data url")
nic = vm.nic[0]
gateway = str(nic.gateway)
- self.debug("GATEWAY: " + gateway)
+ self.debug("Gateway: " + gateway)
user_data_url = 'curl "http://' + gateway + ':80/latest/user-data"'
return user_data_url
- # Creates and verifies the firewall rule
+ # create_and_verify_fw - Creates and verifies (Ingress) firewall rule with a Static NAT rule enabled public IP
def create_and_verify_fw(self, vm, public_ip, network):
- self.debug("Create and verify firewall rule")
+ self.debug("Creating and verifying firewall rule")
self.create_StaticNatRule_For_VM(vm, public_ip, network)
# VSD verification
- self.verify_vsp_floating_ip(network, vm, public_ip.ipaddress)
+ self.verify_vsd_floating_ip(network, vm, public_ip.ipaddress)
fw_rule = self.create_FirewallRule(public_ip, self.test_data["ingress_rule"])
- self.verify_vsp_firewall_rule(fw_rule)
- vm_interface_id = self.get_vm_interface_id(vm)
- pd = self.vsd.get_vm_interface_policydecisions(id=vm_interface_id)
- self.debug(pd)
- egressAcls = pd['egressACLs'][0]['entries']
- gotFirewallPolicy = False
- for acl in egressAcls:
- if acl['destinationPort'] == "22-22":
- gotFirewallPolicy = True
- break
- if not gotFirewallPolicy:
- raise ValueError('No firewall policy decision in vm interface')
+ # VSD verification
+ self.verify_vsd_firewall_rule(fw_rule)
+ self.debug("Successfully created and verified firewall rule")
+
+ # stop_vm - Stops the given VM, and verifies its state
def stop_vm(self, vm):
self.debug("Stoping VM")
- vm.stop(self.apiclient)
- list_vm_response = VirtualMachine.list(self.apiclient,
- id=vm.id)
+ vm.stop(self.api_client)
+ list_vm_response = VirtualMachine.list(self.api_client,
+ id=vm.id
+ )
if isinstance(list_vm_response, list):
vm = list_vm_response[0]
if vm.state != 'Stopped':
- raise Exception("Failed to stop VM (ID: %s) " %
- self.vm.id)
+ raise Exception("Failed to stop VM (ID: %s) " % self.vm.id)
else:
- raise Exception("Invalid response from list_virtual_machines VM (ID: %s) " %
- self.vm.id)
+ raise Exception("Invalid response from list_virtual_machines VM (ID: %s) " % self.vm.id)
+ self.debug("Stopped VM")
+ # install_cloud_set_guest_password_script - Installs the cloud-set-guest-password script from people.apache.org in
+ # the given VM (SSH client)
def install_cloud_set_guest_password_script(self, ssh_client):
- self.debug("GET CLOUD-SET-GUEST-PASSWORD")
+ self.debug("Installing cloud-set-guest-password script")
cmd = "cd /etc/init.d;wget http://people.apache.org/~tsp/cloud-set-guest-password"
result = self.execute_cmd(ssh_client, cmd)
- self.debug("WGET CLOUD-SET-GUEST-PASSWORD: " + result)
+ self.debug("wget file cloud-set-guest-password: " + result)
if "200 OK" not in result:
- self.fail("failed to get file cloud-set-guest-password")
+ self.fail("failed to wget file cloud-set-guest-password")
cmds = ["chmod +x /etc/init.d/cloud-set-guest-password",
"chkconfig --add cloud-set-guest-password"
]
@@ -172,123 +141,145 @@ class TestNuagePasswordReset(nuageTestCase):
result = self.execute_cmd(ssh_client, c)
self.debug("get_set_password_file cmd " + c)
self.debug("get_set_password_file result " + result)
+ self.debug("Installed cloud-set-guest-password script")
@attr(tags=["advanced", "nuagevsp"], required_hardware="true")
def test_nuage_UserDataPasswordReset(self):
"""Test user data and password reset functionality with Nuage VSP SDN plugin
"""
- """
- Validate the following:
- 1) user data
- 2) reset vm password.
-
- Steps:
- 1. Set password enabled to false in the template.
- 2. Create an Isolated network - Test Network (10.1.1.1/24).
- 3. Deploy VM1 in Test Network
- 4. Verify domain,zone subnet, vm.
- 5. create public IP, Create Static Nat rule firewall rule and verify
- 6. SSH to VM should be successful
- 7. verify userdata
- 8. check cloud-set-guest-password exist.
- 9. if cloud-set-guest-password exist.
- 9.1 change template password enabled to true
- 9.2 verify that template is password enbalded
- 9.3 SSH with new password should be successful
- 10. else cloud-set-guest-password does not exist.
- 10.1 get the cloud-set-guest-password file
- 10.2 stop vm
- 10.3 create a new template with password enabled. Verify that template is password enabled.
- 10.4 create vm 2 with new template in Test Network
- 10.5 Verify vm.
- 10.6 create public IP, Create Static Nat rule firewall rule and verify
- 10.7 SSH to VM 2 should be successful
- 11. Reset VM password (VM_1 if guest password file exist. else it is VM2)
- 12 Starting VM and SSH to VM to verify new password
- """
-
- self.debug("TEST USER DATA & PASSWORD RESET ON VM")
+ # 1. Create an Isolated Network with Nuage VSP Isolated Network offering, check if it is successfully created
+ # and is in the "Allocated" state.
+ # 2. Set password enabled to false in the guest VM template.
+ # 3. Deploy a VM in the created Isolated network with user data, check if the Isolated network state is changed
+ # to "Implemented", and both the VM & VR are successfully deployed and are in the "Running" state.
+ # 4. Verify that the guest VM template is not password enabled by checking the deployed VM's password
+ # (password == "password").
+ # 5. SSH into the deployed VM and verify its user data (expected user data == actual user data).
+ # 6. Check for cloud-set-guest-password script in the deployed VM for testing password reset functionality.
+ # 7. if cloud-set-guest-password script does not exist in the deployed VM:
+ # 7.1 Install the cloud-set-guest-password script from people.apache.org in the deployed VM.
+ # 7.2 Stop the deployed VM, and create a new password enabled guest VM template with it.
+ # 7.3 Deploy a new VM in the created Isolated network with the newly created guest VM template,
+ # check if the VM is successfully deployed and is in the "Running" state.
+ # 7.4 Verify that the new guest VM template is password enabled by checking the newly deployed VM's
+ # password (password != "password").
+ # 7.5 SSH into the newly deployed VM for verifying its password.
+ # 8. else cloud-set-guest-password script exists in the deployed VM:
+ # 8.1 Change password enabled to true in the guest VM template.
+ # 8.2 Verify that the guest VM template is password enabled.
+ # 9. Reset VM password, and start the VM.
+ # 10. Verify that the new guest VM template is password enabled by checking the VM's password
+ # (password != "password").
+ # 11. SSH into the VM for verifying its new password after its password reset.
+ # 12. Set password enabled to the default value in the guest VM template.
+ # 13. Delete all the created objects (cleanup).
+
+ self.debug("Testing user data & password reset functionality in an Isolated network...")
+
+ self.debug("Creating an Isolated network...")
+ net_off = self.create_NetworkOffering(self.test_data["nuagevsp"]["isolated_network_offering"])
+ self.network = self.create_Network(net_off)
+ self.validate_Network(self.network, state="Allocated")
+ self.debug("Setting password enabled to false in the guest VM template...")
self.defaultTemplateVal = self.template.passwordenabled
if self.template.passwordenabled:
self.updateTemplate(False)
- self.debug("CREATE AN ISOLATED NETWORK")
- net_off = self.create_NetworkOffering(self.test_data["nuagevsp"]["isolated_network_offering"])
- self.network_1 = self.create_Network(net_off)
- self.cleanup.append(self.network_1)
- expUserData = "hello world vm1"
- userdata = base64.b64encode(expUserData)
- self.test_data["virtual_machine_userdata"]["userdata"] = userdata
- self.debug("DEPLOY VM 1 IN TEST NETWORK")
- # Pass the network and name of the vm type from the testdata with the configuration for the vm
- self.vm_1 = self.create_VM(self.network_1, vm_key="virtual_machine_userdata")
-
- self.vm_1.password = self.test_data["virtual_machine_userdata"]["password"]
- user_data_cmd = self.get_userdata_url(self.vm_1)
+ self.debug("Deploying a VM in the created Isolated network with user data...")
+ expected_user_data = "hello world vm1"
+ user_data = base64.b64encode(expected_user_data)
+ self.test_data["virtual_machine_userdata"]["userdata"] = user_data
+ self.vm_1 = self.create_VM(self.network, testdata=self.test_data["virtual_machine_userdata"])
+ self.validate_Network(self.network, state="Implemented")
+ vr = self.get_Router(self.network)
+ self.check_Router_state(vr, state="Running")
+ self.check_VM_state(self.vm_1, state="Running")
# VSD verification
- self.debug("VERIFY DOMAIN, ZONE, NETWORK , and VM 1")
- self.verify_vsp_network(self.domain.id, self.network_1)
- self.verify_vsp_vm(self.vm_1)
-
- self.debug("CREATE PUBLIC IP, STATIC NAT RULE, FLOATING IP, FIREWALL AND VERIFY")
- public_ip_1 = self.acquire_PublicIPAddress(self.network_1)
- self.create_and_verify_fw(self.vm_1, public_ip_1, self.network_1)
-
- self.debug("SSH TO VM")
+ self.verify_vsd_network(self.domain.id, self.network)
+ self.verify_vsd_router(vr)
+ self.verify_vsd_vm(self.vm_1)
+
+ self.debug("verifying that the guest VM template is not password enabled...")
+ self.debug("VM - %s password - %s !" % (self.vm_1.name, self.vm_1.password))
+ self.assertEqual(self.vm_1.password, self.test_data["virtual_machine_userdata"]["password"],
+ "Password is enabled for the VM (vm_1)"
+ )
+
+ self.debug("SSHing into the VM for verifying its user data...")
+ public_ip_1 = self.acquire_PublicIPAddress(self.network)
+ self.create_and_verify_fw(self.vm_1, public_ip_1, self.network)
ssh = self.ssh_into_VM(self.vm_1, public_ip_1)
-
- self.debug("VERIFY USER DATA")
- self.debug("Get User Data with command: " + user_data_cmd)
- adata = self.execute_cmd(ssh, user_data_cmd)
- actUserData = base64.b64decode(adata)
- self.debug("Response User Data=" + actUserData + ", Expected=" + expUserData)
- self.assertEqual(actUserData, expUserData, "User Data Did Not Match ")
-
- # check /etc/init.d/cloud-set-quest-password
+ user_data_cmd = self.get_userdata_url(self.vm_1)
+ self.debug("Getting user data with command: " + user_data_cmd)
+ actual_user_data = base64.b64decode(self.execute_cmd(ssh, user_data_cmd))
+ self.debug("Actual user data - " + actual_user_data + ", Expected user data - " + expected_user_data)
+ self.assertEqual(actual_user_data, expected_user_data,
+ "Un-expected VM (VM_1) user data"
+ )
+
+ self.debug("Checking for cloud-set-guest-password script in the VM for testing password reset functionality...")
ls_cmd = "ls /etc/init.d/cloud-set-guest-password"
ls_result = self.execute_cmd(ssh, ls_cmd)
ls_result = ls_result.lower()
- self.debug("reponse from ls_cmd: " + ls_result)
+ self.debug("Response from ls_cmd: " + ls_result)
if "no such file" in ls_result:
- self.debug("NO CLOUD-SET_GUEST_PASSWORD FILE. NEED TO GET ONE")
+ self.debug("No cloud-set-guest-password script in the VM")
+ self.debug("Installing the cloud-set-guest-password script from people.apache.org in the VM...")
self.install_cloud_set_guest_password_script(ssh)
+ self.debug("Stopping the VM, and creating a new password enabled guest VM template with it...")
self.stop_vm(self.vm_1)
self.create_template(self.vm_1)
- self.debug("DEPLOY VM 2 IN TEST NETWORK WITH NEW TEMPLATE")
- self.vm_2 = self.create_VM(self.network_1, vm_key="virtual_machine_userdata")
- self.remove_vm2 = True
- self.debug("STARTING VM_2 ")
- vm_2a = self.vm_2.start(self.apiclient)
+
+ self.debug("Deploying a new VM in the created Isolated network with the newly created guest VM template...")
+ self.vm_2 = self.create_VM(self.network, testdata=self.test_data["virtual_machine_userdata"])
+ self.debug("Starting the VM...")
+ vm_2a = self.vm_2.start(self.api_client)
self.vm_2.password = vm_2a.password.strip()
self.vm_2.nic = vm_2a.nic
+
+ # VSD verification
+ self.verify_vsd_vm(self.vm_2)
+
+ self.debug("verifying that the guest VM template is password enabled...")
self.debug("VM - %s password - %s !" % (self.vm_2.name, self.vm_2.password))
- self.assertNotEqual(self.vm_2.password,
- self.test_data["virtual_machine_userdata"]["password"],
- "Password enabled not working. Password same as virtual_machine password "
+ self.assertNotEqual(self.vm_2.password, self.test_data["virtual_machine_userdata"]["password"],
+ "Password is not enabled for the VM"
)
- self.verify_vsp_vm(vm_2a)
- self.debug("GET PUBLIC IP. CREATE AND VERIFIED FIREWALL RULES")
- public_ip_2 = self.acquire_PublicIPAddress(self.network_1)
- self.create_and_verify_fw(self.vm_2, public_ip_2, self.network_1)
+ self.debug("SSHing into the VM for verifying its password...")
+ public_ip_2 = self.acquire_PublicIPAddress(self.network)
+ self.create_and_verify_fw(self.vm_2, public_ip_2, self.network)
self.ssh_into_VM(self.vm_2, public_ip_2)
+
vm_test = self.vm_2
vm_test_public_ip = public_ip_2
-
else:
- self.debug("UPDATE TEMPLATE TO PASSWORD ENABLED")
+ self.debug("Updating the guest VM template to password enabled")
self.updateTemplate(True)
- self.assertEqual(self.template.passwordenabled, True, "Template is not password enabled")
+ self.assertEqual(self.template.passwordenabled, True,
+ "Guest VM template is not password enabled"
+ )
vm_test = self.vm_1
vm_test_public_ip = public_ip_1
- self.debug("RESETTING VM PASSWORD for VM - %s" % vm_test.name)
- vm_test.password = vm_test.resetPassword(self.apiclient)
+ self.debug("Resetting password for VM - %s" % vm_test.name)
+ vm_test.password = vm_test.resetPassword(self.api_client)
self.debug("Password reset to - %s" % vm_test.password)
- self.debug("STARTING VM AND SSH TO VM TO VERIFY NEW PASSWORD")
- vm_test.start(self.apiclient)
- self.debug("VM - %s started!" % vm_test.name)
+
+ self.debug("Starting the VM")
+ vm_test.start(self.api_client)
+
+ self.debug("verifying that the guest VM template is password enabled...")
+ self.debug("VM - %s password - %s !" % (vm_test.name, vm_test.password))
+ self.assertNotEqual(vm_test.password, self.test_data["virtual_machine_userdata"]["password"],
+ "Password is not enabled for the VM"
+ )
+
+ self.debug("SSHing into the VM for verifying its new password after its password reset...")
self.ssh_into_VM(vm_test, vm_test_public_ip)
+
+ self.debug("Setting password enabled to the default value in the guest VM template...")
+ self.updateTemplate(self.defaultTemplateVal)