You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by ta...@apache.org on 2014/12/17 12:57:24 UTC
[4/4] git commit: updated refs/heads/master to 7cf9426
CLOUDSTACK-8074: Fixed maint/test_multiple_ip_ranges.py
Signed-off-by: SrikanteswaraRao Talluri <ta...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/cloudstack/repo
Commit: http://git-wip-us.apache.org/repos/asf/cloudstack/commit/7cf94260
Tree: http://git-wip-us.apache.org/repos/asf/cloudstack/tree/7cf94260
Diff: http://git-wip-us.apache.org/repos/asf/cloudstack/diff/7cf94260
Branch: refs/heads/master
Commit: 7cf94260ee6adc49df97214ea7410dd387be11e5
Parents: 6674f95
Author: Ashutosh K <as...@clogeny.com>
Authored: Tue Dec 16 17:13:05 2014 +0530
Committer: SrikanteswaraRao Talluri <ta...@apache.org>
Committed: Wed Dec 17 17:25:05 2014 +0530
----------------------------------------------------------------------
.../component/maint/test_multiple_ip_ranges.py | 972 ++++++++++---------
1 file changed, 510 insertions(+), 462 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/7cf94260/test/integration/component/maint/test_multiple_ip_ranges.py
----------------------------------------------------------------------
diff --git a/test/integration/component/maint/test_multiple_ip_ranges.py b/test/integration/component/maint/test_multiple_ip_ranges.py
index 982dd7c..f73c470 100644
--- a/test/integration/component/maint/test_multiple_ip_ranges.py
+++ b/test/integration/component/maint/test_multiple_ip_ranges.py
@@ -16,117 +16,72 @@
# under the License.
""" Tests for Multiple IP Ranges feature
"""
-from marvin.cloudstackTestCase import *
-from marvin.cloudstackAPI import *
-from marvin.cloudstackException import CloudstackAPIException
-from marvin.lib.utils import *
-from marvin.lib.base import *
-from marvin.lib.common import *
-# from netaddr import *
+from marvin.cloudstackTestCase import cloudstackTestCase, unittest
+from marvin.lib.utils import cleanup_resources, get_process_status
+from marvin.lib.base import (Account,
+ DiskOffering,
+ VirtualMachine,
+ Router,
+ ServiceOffering,
+ PublicIpRange)
+from marvin.lib.common import (get_domain,
+ get_zone,
+ list_routers,
+ list_hosts,
+ get_pod,
+ get_template)
import netaddr
from nose.plugins.attrib import attr
+from netaddr import IPNetwork, IPAddress
+from marvin.sshClient import SshClient
+import random
-class Services:
- """Test Multiple IP Ranges
- """
- def __init__(self):
- self.services = {
- "account": {
- "email": "test@test.com",
- "firstname": "Test",
- "lastname": "User",
- "username": "test",
- # Random characters are appended for unique
- # username
- "password": "password",
- },
- "service_offering": {
- "name": "Tiny Instance",
- "displaytext": "Tiny Instance",
- "cpunumber": 1,
- "cpuspeed": 200, # in MHz
- "memory": 256, # In MBs
- },
- "disk_offering": {
- "displaytext": "Small Disk",
- "name": "Small Disk",
- "disksize": 1
- },
- "templates": {
- "displaytext": 'Template',
- "name": 'Template',
- "ostype": "CentOS 5.3 (64-bit)",
- "templatefilter": 'self',
- },
- "vlan_ip_range": {
- "startip": "10.147.43.130",
- "endip": "10.147.43.135",
- "netmask": "255.255.255.192",
- "gateway": "10.147.43.129",
- "forvirtualnetwork": "false",
- "vlan": "untagged",
- },
- "server_without_disk": {
- "displayname": "Test VM-No Disk",
- "username": "root",
- "password": "password",
- "hypervisor": 'XenServer',
- },
- "host": {
- "publicport": 22,
- "username": "root", # Host creds for SSH
- "password": "password",
- },
- "ostype": "CentOS 5.3 (64-bit)",
- "sleep": 60,
- "timeout": 10,
- }
class TestMultipleIpRanges(cloudstackTestCase):
+
"""Test Multiple IP Ranges for guest network
"""
@classmethod
def setUpClass(cls):
- cls.testClient = super(TestEgressAfterHostMaintenance, cls).getClsTestClient()
+ cls.testClient = super(TestMultipleIpRanges, cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
cls.dbclient = cls.testClient.getDbConnection()
- cls.services = Services().services
+ cls.testdata = cls.testClient.getParsedTestDataConfig()
# Get Zone, Domain and templates
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.pod = get_pod(cls.api_client, cls.zone.id)
- cls.services['mode'] = cls.zone.networktype
- cls.services["domainid"] = cls.domain.id
- cls.services["zoneid"] = cls.zone.id
+ cls.testdata['mode'] = cls.zone.networktype
+ cls.testdata["domainid"] = cls.domain.id
+ cls.testdata["zoneid"] = cls.zone.id
cls.account = Account.create(
- cls.api_client,
- cls.services["account"],
- domainid=cls.domain.id
- )
- cls.services["account"] = cls.account.name
+ cls.api_client,
+ cls.testdata["account"],
+ domainid=cls.domain.id
+ )
+ cls.testdata["account"] = cls.account.name
cls.disk_offering = DiskOffering.create(
- cls.api_client,
- cls.services["disk_offering"]
- )
+ cls.api_client,
+ cls.testdata["disk_offering"]
+ )
cls.service_offering = ServiceOffering.create(
- cls.api_client,
- cls.services["service_offering"]
- )
+ cls.api_client,
+ cls.testdata["service_offering"]
+ )
cls.template = get_template(
- cls.api_client,
- cls.zone.id,
- cls.services["ostype"]
- )
- cls.services["templates"]["ostypeid"] = cls.template.ostypeid
- cls.services["diskoffering"] = cls.disk_offering.id
+ cls.api_client,
+ cls.zone.id,
+ cls.testdata["ostype"]
+ )
+ cls.testdata["diskoffering"] = cls.disk_offering.id
cls.dc_id = cls.dbclient.execute(
- "select id from data_center where uuid = '%s';" % str(cls.services["zoneid"])
- )
+ "select id from data_center where uuid = '%s';" % str(
+ cls.testdata["zoneid"]))
cls.dc_id = cls.dc_id[0][0]
cls.ids = cls.dbclient.execute(
- "select id from user_ip_address where allocated is null and data_center_id = '%s';" % str(cls.dc_id)
- )
+ "select id from user_ip_address where allocated is null and data_center_id = '%s';" % str(
+ cls.dc_id))
cls.id_list = []
for i in range(len(cls.ids)):
cls.id_list.append(cls.ids[i][0])
@@ -142,45 +97,76 @@ class TestMultipleIpRanges(cloudstackTestCase):
cls.id_list = cls.id_list[:-2]
for id in cls.id_list:
cls.dbclient.execute(
- "update user_ip_address set allocated=now() where id = '%s';" % str(id)
- )
- # Add IP range in the new CIDR
- cls.services["vlan_ip_range"]["zoneid"] = cls.zone.id
- cls.services["vlan_ip_range"]["podid"] = cls.pod.id
+ "update user_ip_address set allocated=now() where id = '%s';" %
+ str(id))
# create new vlan ip range
# Before creating ip range check the zone's network type
- if cls.zone.networktype == 'Basic':
- cls.new_vlan = PublicIpRange.create(cls.api_client, cls.services["vlan_ip_range"])
+ if cls.zone.networktype.lower() == 'basic':
+ cls.new_vlan = cls.createNewVlanRange()
else:
- raise unittest.SkipTest("These tests can be run only on basic zone. So skipping the tests")
+ raise unittest.SkipTest(
+ "These tests can be run only on basic zone.\
+ So skipping the tests")
# Deploy vm in existing subnet if VR is not present
- if cls.vr_state is False :
+ if cls.vr_state is False:
cls.vm_res = VirtualMachine.create(
- cls.api_client,
- cls.services["server_without_disk"],
- templateid=cls.template.id,
- accountid=cls.account.name,
- domainid=cls.services["domainid"],
- zoneid=cls.services["zoneid"],
- serviceofferingid=cls.service_offering.id,
- mode=cls.services["mode"],
- )
+ cls.api_client,
+ cls.testdata["server_without_disk"],
+ templateid=cls.template.id,
+ accountid=cls.account.name,
+ domainid=cls.testdata["domainid"],
+ zoneid=cls.testdata["zoneid"],
+ serviceofferingid=cls.service_offering.id,
+ mode=cls.testdata["mode"],
+ )
cls._cleanup = [
- cls.new_vlan,
- cls.account,
- ]
+ cls.new_vlan,
+ cls.account,
+ ]
return
@classmethod
+ def createNewVlanRange(cls):
+ """ Increment current cidr of vlan range present in network
+ and create new range
+ """
+ publicIpRange = PublicIpRange.list(cls.api_client)
+ cls.startIp = publicIpRange[0].startip
+ cls.endIp = publicIpRange[0].endip
+ cls.gateway = publicIpRange[0].gateway
+ cls.netmask = publicIpRange[0].netmask
+ # Pass ip address and mask length to IPNetwork to findout the CIDR
+ ip = IPNetwork(cls.startIp + "/" + cls.netmask)
+ # Take random increment factor to avoid adding the same vlan ip range
+ # in each test case
+ networkIncrementFactor = random.randint(1,255)
+ new_cidr = ip.__iadd__(networkIncrementFactor)
+ ip2 = IPNetwork(new_cidr)
+ test_nw = ip2.network
+ ip = IPAddress(test_nw)
+ # Add IP range(5 IPs) in the new CIDR
+ test_gateway = ip.__add__(1)
+ test_startIp = ip.__add__(3)
+ test_endIp = ip.__add__(10)
+ # Populating services with new IP range
+ cls.testdata["vlan_ip_range"]["startip"] = test_startIp
+ cls.testdata["vlan_ip_range"]["endip"] = test_endIp
+ cls.testdata["vlan_ip_range"]["gateway"] = test_gateway
+ cls.testdata["vlan_ip_range"]["netmask"] = cls.netmask
+ cls.testdata["vlan_ip_range"]["zoneid"] = cls.zone.id
+ cls.testdata["vlan_ip_range"]["podid"] = cls.pod.id
+
+ return PublicIpRange.create(
+ cls.api_client,
+ cls.testdata["vlan_ip_range"])
+
+ @classmethod
def tearDownClass(cls):
try:
for id in cls.id_list:
cls.dbclient.execute(
- "update user_ip_address set allocated=default where id = '%s';" % str(id)
- )
- # Wait for expunge interval to cleanup VMs
- wait_for_cleanup(cls.api_client, ["expunge.delay", "expunge.interval"])
- time.sleep(30)
+ "update user_ip_address set allocated=default where id = '%s';" %
+ str(id))
# Cleanup resources used
cleanup_resources(cls.api_client, cls._cleanup)
except Exception as e:
@@ -190,48 +176,55 @@ class TestMultipleIpRanges(cloudstackTestCase):
def setUp(self):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
- self.cleanup = [ ]
+ self.cleanup = []
# Deploy guest vm
- try :
+ try:
self.virtual_machine = VirtualMachine.create(
- self.apiclient,
- self.services["server_without_disk"],
- templateid=self.template.id,
- accountid=self.account.name,
- domainid=self.services["domainid"],
- zoneid=self.services["zoneid"],
- serviceofferingid=self.service_offering.id,
- mode=self.services["mode"],
- )
- except Exception as e :
- raise Exception("Warning: Exception during vm deployment: {}".format(e))
+ self.apiclient,
+ self.testdata["server_without_disk"],
+ templateid=self.template.id,
+ accountid=self.account.name,
+ domainid=self.testdata["domainid"],
+ zoneid=self.testdata["zoneid"],
+ serviceofferingid=self.service_offering.id,
+ mode=self.testdata["mode"],
+ )
+ except Exception as e:
+ raise Exception(
+ "Warning: Exception during vm deployment: {}".format(e))
self.vm_response = VirtualMachine.list(
- self.apiclient,
- id=self.virtual_machine.id
- )
+ self.apiclient,
+ id=self.virtual_machine.id
+ )
self.assertEqual(
isinstance(self.vm_response, list),
True,
"Check VM list response returned a valid list"
- )
- self.ip_range = list(netaddr.iter_iprange(unicode(self.services["vlan_ip_range"]["startip"]), unicode(self.services["vlan_ip_range"]["endip"])))
- self.nic_ip = netaddr.IPAddress(unicode(self.vm_response[0].nic[0].ipaddress))
+ )
+ self.ip_range = list(
+ netaddr.iter_iprange(
+ unicode(
+ self.testdata["vlan_ip_range"]["startip"]), unicode(
+ self.testdata["vlan_ip_range"]["endip"])))
+ self.nic_ip = netaddr.IPAddress(
+ unicode(
+ self.vm_response[0].nic[0].ipaddress))
self.debug("vm got {} as ip address".format(self.nic_ip))
self.assertIn(
- self.nic_ip,
- self.ip_range,
- "VM did not get the ip address from the new ip range"
- )
+ self.nic_ip,
+ self.ip_range,
+ "VM did not get the ip address from the new ip range"
+ )
ip_alias = self.dbclient.execute(
- "select ip4_address from nic_ip_alias;"
- )
+ "select ip4_address from nic_ip_alias;"
+ )
self.alias_ip = str(ip_alias[0][0])
self.debug("alias ip : %s" % self.alias_ip)
self.assertNotEqual(
- self.alias_ip,
- None,
- "Error in creating ip alias. Please check MS logs"
- )
+ self.alias_ip,
+ None,
+ "Error in creating ip alias. Please check MS logs"
+ )
self.cleanup.append(self.virtual_machine)
return
@@ -246,35 +239,38 @@ class TestMultipleIpRanges(cloudstackTestCase):
def verify_vlan_range(self, vlan, services):
# compare vlan_list response with configured values
self.assertEqual(
- isinstance(vlan, list),
- True,
- "Check list response returned a valid list"
- )
+ isinstance(vlan, list),
+ True,
+ "Check list response returned a valid list"
+ )
self.assertNotEqual(
- len(vlan),
- 0,
- "check list vlan response"
- )
+ len(vlan),
+ 0,
+ "check list vlan response"
+ )
self.assertEqual(
- str(vlan[0].startip),
- str(services["startip"]),
- "Start IP in vlan ip range is not matched with the configured start ip"
- )
+ str(vlan[0].startip),
+ str(services["startip"]),
+ "Start IP in vlan ip range is not matched with the\
+ configured start ip"
+ )
self.assertEqual(
- str(vlan[0].endip),
- str(services["endip"]),
- "End IP in vlan ip range is not matched with the configured end ip"
- )
+ str(vlan[0].endip),
+ str(services["endip"]),
+ "End IP in vlan ip range is not matched with the configured end ip"
+ )
self.assertEqual(
- str(vlan[0].gateway),
- str(services["gateway"]),
- "gateway in vlan ip range is not matched with the configured gateway"
- )
+ str(vlan[0].gateway),
+ str(services["gateway"]),
+ "gateway in vlan ip range is not matched with the\
+ configured gateway"
+ )
self.assertEqual(
- str(vlan[0].netmask),
- str(services["netmask"]),
- "netmask in vlan ip range is not matched with the configured netmask"
- )
+ str(vlan[0].netmask),
+ str(services["netmask"]),
+ "netmask in vlan ip range is not matched with\
+ the configured netmask"
+ )
return
@attr(tags=["sg"])
@@ -283,14 +279,20 @@ class TestMultipleIpRanges(cloudstackTestCase):
1.Deploy guest vm
2.Verify vm gets the ip address from new cidr
"""
- self.ip_range = list(netaddr.iter_iprange(unicode(self.services["vlan_ip_range"]["startip"]), unicode(self.services["vlan_ip_range"]["endip"])))
- self.nic_ip = netaddr.IPAddress(unicode(self.vm_response[0].nic[0].ipaddress))
+ self.ip_range = list(
+ netaddr.iter_iprange(
+ unicode(
+ self.testdata["vlan_ip_range"]["startip"]), unicode(
+ self.testdata["vlan_ip_range"]["endip"])))
+ self.nic_ip = netaddr.IPAddress(
+ unicode(
+ self.vm_response[0].nic[0].ipaddress))
self.debug("vm got {} as ip address".format(self.nic_ip))
self.assertIn(
- self.nic_ip,
- self.ip_range,
- "VM did not get the ip address from the new ip range"
- )
+ self.nic_ip,
+ self.ip_range,
+ "VM did not get the ip address from the new ip range"
+ )
return
@attr(tags=["sg"])
@@ -300,51 +302,60 @@ class TestMultipleIpRanges(cloudstackTestCase):
2.Verify dns service listens on alias ip in VR
"""
list_router_response = list_routers(
- self.apiclient,
- zoneid=self.zone.id,
- listall=True
- )
+ self.apiclient,
+ zoneid=self.zone.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_router_response, list),
- True,
- "Check list response returns a valid list"
- )
+ isinstance(list_router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
router = list_router_response[0]
hosts = list_hosts(
- self.apiclient,
- zoneid=router.zoneid,
- type='Routing',
- state='Up',
- id=router.hostid
- )
+ self.apiclient,
+ zoneid=router.zoneid,
+ type='Routing',
+ state='Up',
+ id=router.hostid
+ )
self.assertEqual(
- isinstance(hosts, list),
- True,
- "Check list host returns a valid list"
- )
+ isinstance(hosts, list),
+ True,
+ "Check list host returns a valid list"
+ )
host = hosts[0]
self.debug("Router ID: %s, state: %s" % (router.id, router.state))
self.assertEqual(
- router.state,
- 'Running',
- "Check list router response for router state"
- )
+ router.state,
+ 'Running',
+ "Check list router response for router state"
+ )
+
+ port = self.testdata['configurableData']['host']["publicport"]
+ username = self.testdata['configurableData']['host']["username"]
+ password = self.testdata['configurableData']['host']["password"]
+
+ # SSH to host so that host key is saved in first
+ # attempt
+ SshClient(host.ipaddress, port, username, password)
+
proc = self.alias_ip + ":53"
result = get_process_status(
- host.ipaddress,
- self.services['host']["publicport"],
- self.services['host']["username"],
- self.services['host']["password"],
- router.linklocalip,
- "netstat -atnp | grep %s" % proc
- )
+ host.ipaddress,
+ port,
+ username,
+ password,
+ router.linklocalip,
+ "netstat -atnp | grep %s" % proc
+ )
res = str(result)
self.debug("Dns process status on alias ip: %s" % res)
self.assertNotEqual(
- res.find(proc)
- - 1,
- "dnsmasq service is not running on alias ip"
- )
+ res.find(proc)
+ - 1,
+ "dnsmasq service is not running on alias ip"
+ )
return
@attr(tags=["sg"])
@@ -354,51 +365,60 @@ class TestMultipleIpRanges(cloudstackTestCase):
2.Verify password service(socat) listens on alias ip in VR
"""
list_router_response = list_routers(
- self.apiclient,
- zoneid=self.zone.id,
- listall=True
- )
+ self.apiclient,
+ zoneid=self.zone.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_router_response, list),
- True,
- "Check list response returns a valid list"
- )
+ isinstance(list_router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
router = list_router_response[0]
hosts = list_hosts(
- self.apiclient,
- zoneid=router.zoneid,
- type='Routing',
- state='Up',
- id=router.hostid
- )
+ self.apiclient,
+ zoneid=router.zoneid,
+ type='Routing',
+ state='Up',
+ id=router.hostid
+ )
self.assertEqual(
- isinstance(hosts, list),
- True,
- "Check list host returns a valid list"
- )
+ isinstance(hosts, list),
+ True,
+ "Check list host returns a valid list"
+ )
host = hosts[0]
self.debug("Router ID: %s, state: %s" % (router.id, router.state))
self.assertEqual(
- router.state,
- 'Running',
- "Check list router response for router state"
- )
+ router.state,
+ 'Running',
+ "Check list router response for router state"
+ )
+
+ port = self.testdata['configurableData']['host']["publicport"]
+ username = self.testdata['configurableData']['host']["username"]
+ password = self.testdata['configurableData']['host']["password"]
+
+ # SSH to host so that host key is saved in first
+ # attempt
+ SshClient(host.ipaddress, port, username, password)
+
proc = "socat"
result = get_process_status(
- host.ipaddress,
- self.services['host']["publicport"],
- self.services['host']["username"],
- self.services['host']["password"],
- router.linklocalip,
- "netstat -atnp | grep %s" % proc
- )
+ host.ipaddress,
+ port,
+ username,
+ password,
+ router.linklocalip,
+ "netstat -atnp | grep %s" % proc
+ )
res = str(result)
self.debug("password process status on VR: %s" % res)
self.assertNotEqual(
- res.find(self.alias_ip)
- - 1,
- "password service is not running on alias ip"
- )
+ res.find(self.alias_ip)
+ - 1,
+ "password service is not running on alias ip"
+ )
return
@attr(tags=["sg"])
@@ -408,51 +428,60 @@ class TestMultipleIpRanges(cloudstackTestCase):
2.Verify userdata service(apache2) listens on alias ip in VR
"""
list_router_response = list_routers(
- self.apiclient,
- zoneid=self.zone.id,
- listall=True
- )
+ self.apiclient,
+ zoneid=self.zone.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_router_response, list),
- True,
- "Check list response returns a valid list"
- )
+ isinstance(list_router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
router = list_router_response[0]
hosts = list_hosts(
- self.apiclient,
- zoneid=router.zoneid,
- type='Routing',
- state='Up',
- id=router.hostid
- )
+ self.apiclient,
+ zoneid=router.zoneid,
+ type='Routing',
+ state='Up',
+ id=router.hostid
+ )
self.assertEqual(
- isinstance(hosts, list),
- True,
- "Check list host returns a valid list"
- )
+ isinstance(hosts, list),
+ True,
+ "Check list host returns a valid list"
+ )
host = hosts[0]
self.debug("Router ID: %s, state: %s" % (router.id, router.state))
self.assertEqual(
- router.state,
- 'Running',
- "Check list router response for router state"
- )
+ router.state,
+ 'Running',
+ "Check list router response for router state"
+ )
+
+ port = self.testdata['configurableData']['host']["publicport"]
+ username = self.testdata['configurableData']['host']["username"]
+ password = self.testdata['configurableData']['host']["password"]
+
+ # SSH to host so that host key is saved in first
+ # attempt
+ SshClient(host.ipaddress, port, username, password)
+
proc = "apache2"
result = get_process_status(
- host.ipaddress,
- self.services['host']["publicport"],
- self.services['host']["username"],
- self.services['host']["password"],
- router.linklocalip,
- "netstat -atnp | grep %s" % proc
- )
+ host.ipaddress,
+ port,
+ username,
+ password,
+ router.linklocalip,
+ "netstat -atnp | grep %s" % proc
+ )
res = str(result)
self.debug("userdata process status on VR: %s" % res)
self.assertNotEqual(
- res.find(self.alias_ip + ":80 ")
- - 1,
- "password service is not running on alias ip"
- )
+ res.find(self.alias_ip + ":80 ")
+ - 1,
+ "password service is not running on alias ip"
+ )
return
@attr(tags=["sg"])
@@ -464,93 +493,94 @@ class TestMultipleIpRanges(cloudstackTestCase):
4.Verify ip alias removal after vm expunge
"""
list_router_response = list_routers(
- self.apiclient,
- zoneid=self.zone.id,
- listall=True
- )
+ self.apiclient,
+ zoneid=self.zone.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_router_response, list),
- True,
- "Check list response returns a valid list"
- )
+ isinstance(list_router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
router = list_router_response[0]
hosts = list_hosts(
- self.apiclient,
- zoneid=router.zoneid,
- type='Routing',
- state='Up',
- id=router.hostid
- )
+ self.apiclient,
+ zoneid=router.zoneid,
+ type='Routing',
+ state='Up',
+ id=router.hostid
+ )
self.assertEqual(
- isinstance(hosts, list),
- True,
- "Check list host returns a valid list"
- )
+ isinstance(hosts, list),
+ True,
+ "Check list host returns a valid list"
+ )
host = hosts[0]
self.debug("Router ID: %s, state: %s" % (router.id, router.state))
self.assertEqual(
- router.state,
- 'Running',
- "Check list router response for router state"
- )
+ router.state,
+ 'Running',
+ "Check list router response for router state"
+ )
+
+ port = self.testdata['configurableData']['host']["publicport"]
+ username = self.testdata['configurableData']['host']["username"]
+ password = self.testdata['configurableData']['host']["password"]
+
+ # SSH to host so that host key is saved in first
+ # attempt
+ SshClient(host.ipaddress, port, username, password)
+
proc = "ip addr show eth0"
result = get_process_status(
- host.ipaddress,
- self.services['host']["publicport"],
- self.services['host']["username"],
- self.services['host']["password"],
- router.linklocalip,
- proc
- )
+ host.ipaddress,
+ port,
+ username,
+ password,
+ router.linklocalip,
+ proc
+ )
res = str(result)
self.debug("ip alias configuration on VR: %s" % res)
self.assertNotEqual(
- res.find(self.alias_ip)
- - 1,
- "ip alias is not created on VR eth0"
- )
+ res.find(self.alias_ip)
+ - 1,
+ "ip alias is not created on VR eth0"
+ )
self.virtual_machine.delete(self.apiclient)
- expunge_del = Configurations.list(
- self.apiclient,
- name='expunge.delay'
- )
- expunge_int = Configurations.list(
- self.apiclient,
- name='expunge.interval'
- )
- wait_time = int(expunge_del[0].value) + int(expunge_int[0].value) + int(30)
- self.debug("Waiting for {} seconds for the vm to expunge".format(wait_time))
- # wait for the vm to expunge
- time.sleep(wait_time)
- self.debug("Verify that expunging the last vm in the CIDR should delete the ip alias from VR")
+ self.debug(
+ "Verify that expunging the last vm in the CIDR should\
+ delete the ip alias from VR")
ip_alias2 = self.dbclient.execute(
- "select ip4_address from nic_ip_alias;"
- )
+ "select ip4_address from nic_ip_alias;"
+ )
self.assertEqual(
- isinstance(ip_alias2, list),
- True,
- "Error in sql query"
- )
+ isinstance(ip_alias2, list),
+ True,
+ "Error in sql query"
+ )
self.assertEqual(
- len(ip_alias2),
- 0,
- "Failure in clearing ip alias entry from cloud db"
- )
+ len(ip_alias2),
+ 0,
+ "Failure in clearing ip alias entry from cloud db"
+ )
+
proc = "ip addr show eth0"
result = get_process_status(
- host.ipaddress,
- self.services['host']["publicport"],
- self.services['host']["username"],
- self.services['host']["password"],
- router.linklocalip,
- proc
- )
+ host.ipaddress,
+ port,
+ username,
+ password,
+ router.linklocalip,
+ proc
+ )
res = str(result)
self.assertEqual(
- res.find(self.alias_ip),
- - 1,
- "Failed to clean up ip alias from VR even after last vm expunge in the CIDR"
- )
+ res.find(
+ self.alias_ip),
+ - 1,
+ "Failed to clean up ip alias from VR even after\
+ last vm expunge in the CIDR")
self.debug("IP alias got deleted from VR successfully.")
self.cleanup.remove(self.virtual_machine)
return
@@ -564,86 +594,95 @@ class TestMultipleIpRanges(cloudstackTestCase):
4.Verify ip alias on VR
"""
list_router_response = list_routers(
- self.apiclient,
- zoneid=self.zone.id,
- listall=True
- )
+ self.apiclient,
+ zoneid=self.zone.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_router_response, list),
- True,
- "Check list response returns a valid list"
- )
+ isinstance(list_router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
router = list_router_response[0]
hosts = list_hosts(
- self.apiclient,
- zoneid=router.zoneid,
- type='Routing',
- state='Up',
- id=router.hostid
- )
+ self.apiclient,
+ zoneid=router.zoneid,
+ type='Routing',
+ state='Up',
+ id=router.hostid
+ )
self.assertEqual(
- isinstance(hosts, list),
- True,
- "Check list host returns a valid list"
- )
+ isinstance(hosts, list),
+ True,
+ "Check list host returns a valid list"
+ )
host = hosts[0]
self.debug("Router ID: %s, state: %s" % (router.id, router.state))
self.assertEqual(
- router.state,
- 'Running',
- "Check list router response for router state"
- )
+ router.state,
+ 'Running',
+ "Check list router response for router state"
+ )
+
+ port = self.testdata['configurableData']['host']["publicport"]
+ username = self.testdata['configurableData']['host']["username"]
+ password = self.testdata['configurableData']['host']["password"]
+
+ # SSH to host so that host key is saved in first
+ # attempt
+ SshClient(host.ipaddress, port, username, password)
+
proc = "ip addr show eth0"
result = get_process_status(
- host.ipaddress,
- self.services['host']["publicport"],
- self.services['host']["username"],
- self.services['host']["password"],
- router.linklocalip,
- proc
- )
+ host.ipaddress,
+ port,
+ username,
+ password,
+ router.linklocalip,
+ proc
+ )
res = str(result)
self.debug("ip alias configuration on VR: %s" % res)
self.assertNotEqual(
- res.find(self.alias_ip)
- - 1,
- "ip alias is not created on VR eth0"
- )
+ res.find(self.alias_ip)
+ - 1,
+ "ip alias is not created on VR eth0"
+ )
resp = Router.reboot(
- self.apiclient,
- router.id
- )
+ self.apiclient,
+ router.id
+ )
self.debug("Reboot router api response: %s" % resp)
list_router_response = list_routers(
- self.apiclient,
- zoneid=self.zone.id,
- listall=True
- )
+ self.apiclient,
+ zoneid=self.zone.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_router_response, list),
- True,
- "Check list response returns a valid list"
- )
+ isinstance(list_router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
router = list_router_response[0]
self.assertEqual(
- router.state,
- 'Running',
- "Router is not in running state after reboot"
- )
+ router.state,
+ 'Running',
+ "Router is not in running state after reboot"
+ )
result = get_process_status(
- host.ipaddress,
- self.services['host']["publicport"],
- self.services['host']["username"],
- self.services['host']["password"],
- router.linklocalip,
- proc
- )
+ host.ipaddress,
+ port,
+ username,
+ password,
+ router.linklocalip,
+ proc
+ )
res = str(result)
self.assertNotEqual(
- res.find(self.alias_ip),
- - 1,
- "IP alias not present on VR after VR reboot"
- )
+ res.find(self.alias_ip),
+ - 1,
+ "IP alias not present on VR after VR reboot"
+ )
return
@attr(tags=["sg"])
@@ -655,90 +694,99 @@ class TestMultipleIpRanges(cloudstackTestCase):
4.Verify ip alias on VR
"""
list_router_response = list_routers(
- self.apiclient,
- zoneid=self.zone.id,
- listall=True
- )
+ self.apiclient,
+ zoneid=self.zone.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_router_response, list),
- True,
- "Check list response returns a valid list"
- )
+ isinstance(list_router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
router = list_router_response[0]
hosts = list_hosts(
- self.apiclient,
- zoneid=router.zoneid,
- type='Routing',
- state='Up',
- id=router.hostid
- )
+ self.apiclient,
+ zoneid=router.zoneid,
+ type='Routing',
+ state='Up',
+ id=router.hostid
+ )
self.assertEqual(
- isinstance(hosts, list),
- True,
- "Check list host returns a valid list"
- )
+ isinstance(hosts, list),
+ True,
+ "Check list host returns a valid list"
+ )
host = hosts[0]
self.debug("Router ID: %s, state: %s" % (router.id, router.state))
self.assertEqual(
- router.state,
- 'Running',
- "Check list router response for router state"
- )
+ router.state,
+ 'Running',
+ "Check list router response for router state"
+ )
+
+ port = self.testdata['configurableData']['host']["publicport"]
+ username = self.testdata['configurableData']['host']["username"]
+ password = self.testdata['configurableData']['host']["password"]
+
+ # SSH to host so that host key is saved in first
+ # attempt
+ SshClient(host.ipaddress, port, username, password)
+
proc = "ip addr show eth0"
result = get_process_status(
- host.ipaddress,
- self.services['host']["publicport"],
- self.services['host']["username"],
- self.services['host']["password"],
- router.linklocalip,
- proc
- )
+ host.ipaddress,
+ port,
+ username,
+ password,
+ router.linklocalip,
+ proc
+ )
res = str(result)
self.debug("ip alias configuration on VR: %s" % res)
self.assertNotEqual(
- res.find(self.alias_ip)
- - 1,
- "ip alias is not created on VR eth0"
- )
+ res.find(self.alias_ip)
+ - 1,
+ "ip alias is not created on VR eth0"
+ )
self.debug("Stopping VR")
- stop_res = Router.stop(
- self.apiclient,
- router.id,
- )
+ Router.stop(
+ self.apiclient,
+ router.id,
+ )
self.debug("Starting VR")
- start_res = Router.start(
- self.apiclient,
- router.id
- )
+ Router.start(
+ self.apiclient,
+ router.id
+ )
list_router_response = list_routers(
- self.apiclient,
- zoneid=self.zone.id,
- listall=True
- )
+ self.apiclient,
+ zoneid=self.zone.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_router_response, list),
- True,
- "Check list response returns a valid list"
- )
+ isinstance(list_router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
router = list_router_response[0]
self.assertEqual(
- router.state,
- 'Running',
- "Router is not in running state after reboot"
- )
+ router.state,
+ 'Running',
+ "Router is not in running state after reboot"
+ )
self.debug("VR is up and Running")
result = get_process_status(
- host.ipaddress,
- self.services['host']["publicport"],
- self.services['host']["username"],
- self.services['host']["password"],
- router.linklocalip,
- proc
- )
+ host.ipaddress,
+ port,
+ username,
+ password,
+ router.linklocalip,
+ proc
+ )
res = str(result)
self.assertNotEqual(
- res.find(self.alias_ip),
- - 1,
- "IP alias not present on VR after VR stop and start"
- )
+ res.find(self.alias_ip),
+ - 1,
+ "IP alias not present on VR after VR stop and start"
+ )
return