diff --git a/test/integration/component/maint/test_vpc_host_maintenance.py b/test/integration/component/maint/test_vpc_host_maintenance.py index fbc1bb484e4..702746918cf 100644 --- a/test/integration/component/maint/test_vpc_host_maintenance.py +++ b/test/integration/component/maint/test_vpc_host_maintenance.py @@ -36,6 +36,7 @@ from marvin.lib.common import (get_domain, get_zone, get_template, wait_for_cleanup) +import time class Services: @@ -190,7 +191,7 @@ class Services: "ostype": 'CentOS 5.3 (64-bit)', # Cent OS 5.3 (64 bit) "sleep": 60, - "timeout": 10, + "timeout": 30, "mode": 'advanced' } @@ -376,7 +377,7 @@ class TestVMLifeCycleHostmaintenance(cloudstackTestCase): self.router = routers[0] if self.router.state == "Running": - Router.stop(self.apiclient, id=self.router.id) + self.debug("Verified that the Router is in Running State") self.cleanup = [] return @@ -448,6 +449,9 @@ class TestVMLifeCycleHostmaintenance(cloudstackTestCase): # 2. VMs present on the above host should successfully migrate to the # other host present in the cluster + Host.update(self.apiclient, id=self.hosts[0].id, hosttags="hosttag1,hosttag2") + Host.update(self.apiclient, id=self.hosts[1].id, hosttags="hosttag1,hosttag2") + self.validate_vm_deployment() self.debug("Stop the host on which the VPC virtual router is running") try: @@ -468,12 +472,34 @@ class TestVMLifeCycleHostmaintenance(cloudstackTestCase): True, "List routers shall return a valid VPCVR for account" ) - for router in routers: - self.assertEqual( - router.state, - "Running", - "Router state should be running after migration" - ) + + router = routers[0] + + try: + + timeout = self.services["timeout"] + self.debug("Timeout Value %d : " % timeout) + while True: + list_router_response = Router.list( + self.apiclient, + id = router.id, + state = "Running" + ) + + if list_router_response is not None: + break + elif timeout == 0: + raise Exception("Router state should be running after migration") + + time.sleep(self.services["sleep"]) + timeout = timeout - 1 + self.debug("Waiting for %d seconds - %d tries left" % (self.services["sleep"],timeout)) + + self.debug("Verified that the Router is in Running State") + + except Exception as e: + self.fail("Failed to find the Router in Running state %s " % e) + vms = VirtualMachine.list( self.apiclient, account=self.account.name, @@ -488,7 +514,7 @@ class TestVMLifeCycleHostmaintenance(cloudstackTestCase): for vm in vms: self.assertEqual( vm.state, - "Ruuning", + "Running", "Vm state should be running after migration" ) return @@ -506,12 +532,51 @@ class TestVMLifeCycleHostmaintenance(cloudstackTestCase): # 1. Successfully cancel the Maintenance mode on the host. # 2. Migrate the VMs back successfully onto the host. # 3. Check that the network connectivity exists with the migrated VMs. + + try: + timeout = self.services["timeout"] + while True: + list_host_response = Host.list( + self.apiclient, + id=self.vpcvr.hostid, + resourcestate="Maintenance") + + if list_host_response is not None: + break + elif timeout == 0: + raise Exception("Failed to list the Host in Maintenance State") + + time.sleep(self.services["sleep"]) + timeout = timeout - 1 + + self.debug("Verified that the Host is in Maintenance State") + + except: + self.fail("Failed to find the Host in maintenance state") self.debug("Cancel host maintenence on which the VPCVR is running") try: Host.cancelMaintenance(self.apiclient, id=self.vpcvr.hostid) + + timeout = self.services["timeout"] + while True: + list_host_response = Host.list( + self.apiclient, + id=self.vpcvr.hostid, + state="Up") + + if list_host_response is not None: + break + elif timeout == 0: + raise Exception("Failed to list the Host in Up State after Canceling Maintenance Mode") + + time.sleep(self.services["sleep"]) + timeout = timeout - 1 + + self.debug("Verified that the Host is in Up State after Canceling Maintenance Mode") + except Exception as e: - self.fail("Failed to enable maintenance mode on host: %s" % e) + self.fail("Failed to cancel maintenance mode on host: %s" % e) self.debug( "Migrating the instances back to the host: %s" % @@ -555,6 +620,27 @@ class TestVMLifeCycleHostmaintenance(cloudstackTestCase): # 2. Network connectivity to all the VMs on the host should not be # effected due to reconnection. + try: + timeout = self.services["timeout"] + while True: + list_host_response = Host.list( + self.apiclient, + id=self.vpcvr.hostid, + resourcestate="Enabled") + + if list_host_response is not None: + break + elif timeout == 0: + raise Exception("Failed to list the Host in Up State") + + time.sleep(self.services["sleep"]) + timeout = timeout - 1 + + self.debug("Verified that the Host is in Up State") + + except: + self.fail("Failed to find the Host in Up State") + self.debug("Reconnecting the host where VPC VR is running") try: Host.reconnect(self.apiclient, id=self.vpcvr.hostid) diff --git a/test/integration/component/test_eip_elb.py b/test/integration/component/test_eip_elb.py index c35ea9bd240..76987b94254 100644 --- a/test/integration/component/test_eip_elb.py +++ b/test/integration/component/test_eip_elb.py @@ -5,9 +5,9 @@ # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at -# +# # http://www.apache.org/licenses/LICENSE-2.0 -# +# # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY @@ -17,75 +17,83 @@ """ P1 tests for elastic load balancing and elastic IP """ -#Import Local Modules -import marvin +# Import Local Modules from nose.plugins.attrib import attr -from marvin.cloudstackTestCase import * -from marvin.cloudstackAPI import * -from marvin.lib.utils import * -from marvin.lib.base import * -from marvin.lib.common import * +from marvin.cloudstackTestCase import cloudstackTestCase, unittest +from marvin.cloudstackAPI import (authorizeSecurityGroupIngress, + disassociateIpAddress, + deleteLoadBalancerRule) +from marvin.lib.utils import cleanup_resources +from marvin.lib.base import (Account, + PublicIPAddress, + VirtualMachine, + Network, + LoadBalancerRule, + SecurityGroup, + ServiceOffering, + StaticNATRule, + PublicIpRange) +from marvin.lib.common import (get_zone, + get_domain, + get_template) from marvin.sshClient import SshClient -import datetime +import time class Services: + """Test elastic load balancing and elastic IP """ 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": 100, # in MHz - "memory": 128, # In MBs - }, - "lbrule": { - "name": "SSH", - "alg": "roundrobin", - # Algorithm used for load balancing - "privateport": 22, - "publicport": 22, - "openfirewall": False, - }, - "natrule": { - "privateport": 22, - "publicport": 22, - "protocol": "TCP" - }, - "virtual_machine": { - "displayname": "Test VM", - "username": "root", - "password": "password", - "ssh_port": 22, - "hypervisor": 'XenServer', - # Hypervisor type should be same as - # hypervisor type of cluster - "privateport": 22, - "publicport": 22, - "protocol": 'TCP', - }, - "netscaler": { - "ipaddress": '10.147.40.100', - "username": 'nsroot', - "password": 'nsroot' - }, - "ostype": 'CentOS 5.3 (64-bit)', - # Cent OS 5.3 (64 bit) - "sleep": 60, - "timeout": 10, - } + "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": 100, # in MHz + "memory": 128, # In MBs + }, + "lbrule": { + "name": "SSH", + "alg": "roundrobin", + # Algorithm used for load balancing + "privateport": 22, + "publicport": 22, + "openfirewall": False, + }, + "natrule": { + "privateport": 22, + "publicport": 22, + "protocol": "TCP" + }, + "virtual_machine": { + "displayname": "Test VM", + "username": "root", + "password": "password", + "ssh_port": 22, + "hypervisor": 'XenServer', + # Hypervisor type should be same as + # hypervisor type of cluster + "privateport": 22, + "publicport": 22, + "protocol": 'TCP', + }, + "ostype": 'CentOS 5.3 (64-bit)', + # Cent OS 5.3 (64 bit) + "sleep": 60, + "timeout": 10, + } + class TestEIP(cloudstackTestCase): @@ -95,74 +103,85 @@ class TestEIP(cloudstackTestCase): cls.api_client = cls.testClient.getApiClient() cls.services = Services().services + + try: + cls.services["netscaler"] = cls.config.__dict__[ + "netscalerDevice"].__dict__ + except KeyError: + raise unittest.SkipTest("Please make sure you have included netscalerDevice\ + dict in your config file (keys - ipaddress, username,\ + password") + except Exception as e: + raise unittest.SkipTest(e) + # Get Zone, Domain and templates cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.services['mode'] = cls.zone.networktype cls.template = get_template( - cls.api_client, - cls.zone.id, - cls.services["ostype"] - ) + cls.api_client, + cls.zone.id, + cls.services["ostype"] + ) cls.services["virtual_machine"]["zoneid"] = cls.zone.id cls.services["virtual_machine"]["template"] = cls.template.id cls.service_offering = ServiceOffering.create( - cls.api_client, - cls.services["service_offering"] - ) + cls.api_client, + cls.services["service_offering"] + ) cls.account = Account.create( - cls.api_client, - cls.services["account"], - admin=True, - domainid=cls.domain.id - ) + cls.api_client, + cls.services["account"], + admin=True, + domainid=cls.domain.id + ) # Spawn an instance cls.virtual_machine = VirtualMachine.create( - cls.api_client, - cls.services["virtual_machine"], - accountid=cls.account.name, - domainid=cls.account.domainid, - serviceofferingid=cls.service_offering.id - ) + cls.api_client, + cls.services["virtual_machine"], + accountid=cls.account.name, + domainid=cls.account.domainid, + serviceofferingid=cls.service_offering.id + ) networks = Network.list( - cls.api_client, - zoneid=cls.zone.id, - listall=True - ) + cls.api_client, + zoneid=cls.zone.id, + listall=True + ) if isinstance(networks, list): # Basic zone has only one network i.e Basic network cls.guest_network = networks[0] else: raise Exception( "List networks returned empty response for zone: %s" % - cls.zone.id) + cls.zone.id) ip_addrs = PublicIPAddress.list( - cls.api_client, - associatednetworkid=cls.guest_network.id, - isstaticnat=True, - account=cls.account.name, - domainid=cls.account.domainid, - listall=True - ) + cls.api_client, + associatednetworkid=cls.guest_network.id, + isstaticnat=True, + account=cls.account.name, + domainid=cls.account.domainid, + listall=True + ) if isinstance(ip_addrs, list): cls.source_nat = ip_addrs[0] - print "source_nat ipaddress : ", cls.source_nat + print "source_nat ipaddress : ", cls.source_nat.ipaddress else: raise Exception( - "No Source NAT IP found for guest network: %s" % - cls.guest_network.id) + "No Source NAT IP found for guest network: %s" % + cls.guest_network.id) cls._cleanup = [ - cls.account, - cls.service_offering, - ] + cls.account, + cls.service_offering, + ] return @classmethod def tearDownClass(cls): try: - #Cleanup resources used + # Cleanup resources used cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) @@ -176,19 +195,17 @@ class TestEIP(cloudstackTestCase): def tearDown(self): try: - #Clean up, terminate the created network offerings + # Clean up, terminate the created network offerings cleanup_resources(self.apiclient, self.cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return - - @attr(tags = ["eip"]) + @attr(tags=["eip"]) def test_01_eip_by_deploying_instance(self): """Test EIP by deploying an instance """ - # Validate the following # 1. Instance gets an IP from GUEST IP range. # 2. One IP from EIP pool is taken and configured on NS @@ -200,62 +217,57 @@ class TestEIP(cloudstackTestCase): self.debug("Fetching public network IP range for public network") ip_ranges = PublicIpRange.list( - self.apiclient, - zoneid=self.zone.id, - forvirtualnetwork=True - ) + self.apiclient, + zoneid=self.zone.id, + forvirtualnetwork=True + ) self.assertEqual( - isinstance(ip_ranges, list), - True, - "Public IP range should return a valid range" - ) + isinstance(ip_ranges, list), + True, + "Public IP range should return a valid range" + ) # Guest network can have multiple IP ranges. In that case, split IP # address and then compare the values for ip_range in ip_ranges: self.debug("IP range: %s - %s" % ( - ip_range.startip, - ip_range.endip - )) + ip_range.startip, + ip_range.endip + )) start_ip_list = ip_range.startip.split(".") end_ip_list = ip_range.endip.split(".") source_nat_list = self.source_nat.ipaddress.split(".") - flag = True - for i in range(3): - if int(start_ip_list[i]) != int(source_nat_list[i]): - flag = False - self.assertGreaterEqual( int(source_nat_list[3]), int(start_ip_list[3]), "The NAT should be greater/equal to start IP of guest network" - ) + ) self.assertLessEqual( int(source_nat_list[3]), int(end_ip_list[3]), "The NAT should be less/equal to start IP of guest network" - ) + ) # Verify listSecurity groups response security_groups = SecurityGroup.list( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid + ) self.assertEqual( - isinstance(security_groups, list), - True, - "Check for list security groups response" - ) + isinstance(security_groups, list), + True, + "Check for list security groups response" + ) self.assertEqual( - len(security_groups), - 1, - "Check List Security groups response" - ) + len(security_groups), + 1, + "Check List Security groups response" + ) self.debug("List Security groups response: %s" % - str(security_groups)) + str(security_groups)) security_group = security_groups[0] @@ -271,7 +283,7 @@ class TestEIP(cloudstackTestCase): cmd.endport = 22 cmd.cidrlist = '0.0.0.0/0' self.apiclient.authorizeSecurityGroupIngress(cmd) -#COMMENTED: +# COMMENTED: # try: # self.debug("SSH into VM: %s" % self.virtual_machine.ssh_ip) # ssh = self.virtual_machine.get_ssh_client( @@ -282,80 +294,77 @@ class TestEIP(cloudstackTestCase): # ) # Fetch details from user_ip_address table in database self.debug( - "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \ - % self.source_nat.ipaddress) + "select is_system, one_to_one_nat from user_ip_address\ + where public_ip_address='%s';" % + self.source_nat.ipaddress) qresultset = self.dbclient.execute( - "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \ - % self.source_nat.ipaddress - ) + "select is_system, one_to_one_nat from user_ip_address\ + where public_ip_address='%s';" % + self.source_nat.ipaddress) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "Check DB Query result set" - ) + len(qresultset), + 0, + "Check DB Query result set" + ) qresult = qresultset[0] self.assertEqual( - qresult[0], - 1, - "user_ip_address.is_system value should be 1 for static NAT" - ) + qresult[0], + 1, + "user_ip_address.is_system value should be 1 for static NAT" + ) self.assertEqual( - qresult[1], - 1, - "user_ip_address.one_to_one_nat value should be 1 for static NAT" - ) + qresult[1], + 1, + "user_ip_address.one_to_one_nat value should be 1 for static NAT" + ) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) - try: - ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - 22, - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], - ) - self.debug("command: show ip") - res = ssh_client.execute("show ip") - result = str(res) - self.debug("Output: %s" % result) + self.services["netscaler"]["ipaddress"]) - self.assertEqual( - result.count(self.source_nat.ipaddress), - 1, - "One IP from EIP pool should be taken and configured on NS" - ) + ssh_client = SshClient( + self.services["netscaler"]["ipaddress"], + 22, + self.services["netscaler"]["username"], + self.services["netscaler"]["password"], + ) + self.debug("command: show ip") + res = ssh_client.execute("show ip") + result = str(res) + self.debug("Output: %s" % result) - self.debug("Command:show inat") - res = ssh_client.execute("show inat") + self.assertEqual( + result.count(self.source_nat.ipaddress), + 1, + "One IP from EIP pool should be taken and configured on NS" + ) - result = str(res) - self.debug("Output: %s" % result) + self.debug("Command:show inat") + res = ssh_client.execute("show inat") - self.assertEqual( - result.count("NAME: Cloud-Inat-%s" % self.source_nat.ipaddress), - 1, - "User source IP should be enabled for INAT service" - ) + result = str(res) + self.debug("Output: %s" % result) - except Exception as e: - self.fail("SSH Access failed for %s: %s" % \ - (self.services["netscaler"]["ipaddress"], e)) + self.assertEqual( + result.count( + "NAME: Cloud-Inat-%s" % + self.source_nat.ipaddress), + 1, + "User source IP should be enabled for INAT service") return - @attr(tags = ["eip"]) + @attr(tags=["eip"]) def test_02_acquire_ip_enable_static_nat(self): """Test associate new IP and enable static NAT for new IP and the VM """ - # Validate the following # 1. user_ip_address.is_system = 0 & user_ip_address.one_to_one_nat=1 # 2. releases default EIP whose user_ip_address.is_system=1 @@ -369,84 +378,86 @@ class TestEIP(cloudstackTestCase): self.debug("Acquiring new IP for network: %s" % self.guest_network.id) public_ip = PublicIPAddress.create( - self.apiclient, - accountid=self.account.name, - zoneid=self.zone.id, - domainid=self.account.domainid, - services=self.services["virtual_machine"] - ) + self.apiclient, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + services=self.services["virtual_machine"] + ) self.debug("IP address: %s is acquired by network: %s" % ( - public_ip.ipaddress, - self.guest_network.id)) + public_ip.ipaddress.ipaddress, + self.guest_network.id)) self.debug("Enabling static NAT for IP Address: %s" % - public_ip.ipaddress) + public_ip.ipaddress.ipaddress) StaticNATRule.enable( - self.apiclient, - ipaddressid=public_ip.ipaddress.id, - virtualmachineid=self.virtual_machine.id - ) + self.apiclient, + ipaddressid=public_ip.ipaddress.id, + virtualmachineid=self.virtual_machine.id + ) # Fetch details from user_ip_address table in database self.debug( - "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \ - % public_ip.ipaddress) + "select is_system, one_to_one_nat from user_ip_address\ + where public_ip_address='%s';" % + public_ip.ipaddress.ipaddress) qresultset = self.dbclient.execute( - "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \ - % public_ip.ipaddress - ) + "select is_system, one_to_one_nat from user_ip_address\ + where public_ip_address='%s';" % + public_ip.ipaddress.ipaddress) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "Check DB Query result set" - ) + len(qresultset), + 0, + "Check DB Query result set" + ) qresult = qresultset[0] self.assertEqual( - qresult[0], - 0, - "user_ip_address.is_system value should be 0 for new IP" - ) + qresult[0], + 0, + "user_ip_address.is_system value should be 0 for new IP" + ) self.assertEqual( - qresult[1], - 1, - "user_ip_address.one_to_one_nat value should be 1 for static NAT" - ) + qresult[1], + 1, + "user_ip_address.one_to_one_nat value should be 1 for static NAT" + ) self.debug( - "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \ - % self.source_nat.ipaddress) + "select is_system, one_to_one_nat from user_ip_address\ + where public_ip_address='%s';" % + self.source_nat.ipaddress) qresultset = self.dbclient.execute( - "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \ - % self.source_nat.ipaddress - ) + "select is_system, one_to_one_nat from user_ip_address\ + where public_ip_address='%s';" % + self.source_nat.ipaddress) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "Check DB Query result set" - ) + len(qresultset), + 0, + "Check DB Query result set" + ) qresult = qresultset[0] self.assertEqual( - qresult[0], - 0, - "user_ip_address.is_system value should be 0 old source NAT" - ) + qresult[0], + 0, + "user_ip_address.is_system value should be 0 old source NAT" + ) # try: # self.debug("SSH into VM: %s" % public_ip.ipaddress) @@ -458,24 +469,24 @@ class TestEIP(cloudstackTestCase): # ) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler"]["ipaddress"]) try: ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - 22, - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], - ) + self.services["netscaler"]["ipaddress"], + 22, + self.services["netscaler"]["username"], + self.services["netscaler"]["password"], + ) self.debug("command: show ip") res = ssh_client.execute("show ip") result = str(res) self.debug("Output: %s" % result) self.assertEqual( - result.count(public_ip.ipaddress), - 1, - "One IP from EIP pool should be taken and configured on NS" - ) + result.count(public_ip.ipaddress.ipaddress), + 1, + "One IP from EIP pool should be taken and configured on NS" + ) self.debug("Command:show inat") res = ssh_client.execute("show inat") @@ -484,22 +495,22 @@ class TestEIP(cloudstackTestCase): self.debug("Output: %s" % result) self.assertEqual( - result.count("NAME: Cloud-Inat-%s" % public_ip.ipaddress), - 1, - "User source IP should be enabled for INAT service" - ) + result.count( + "NAME: Cloud-Inat-%s" % + public_ip.ipaddress.ipaddress), + 1, + "User source IP should be enabled for INAT service") except Exception as e: - self.fail("SSH Access failed for %s: %s" % \ + self.fail("SSH Access failed for %s: %s" % (self.services["netscaler"]["ipaddress"], e)) return - @attr(tags = ["eip"]) + @attr(tags=["eip"]) def test_03_disable_static_nat(self): """Test disable static NAT and release EIP acquired """ - # Validate the following # 1. Disable static NAT. Disables one-to-one NAT and releases EIP # whose user_ip_address.is_system=0 @@ -513,53 +524,56 @@ class TestEIP(cloudstackTestCase): self.debug( "Fetching static NAT for VM: %s" % self.virtual_machine.name) ip_addrs = PublicIPAddress.list( - self.api_client, - associatednetworkid=self.guest_network.id, - isstaticnat=True, - account=self.account.name, - domainid=self.account.domainid, - listall=True - ) + self.api_client, + associatednetworkid=self.guest_network.id, + isstaticnat=True, + account=self.account.name, + domainid=self.account.domainid, + listall=True + ) self.assertEqual( - isinstance(ip_addrs, list), - True, - "List Public IP address should return valid IP address for network" - ) + isinstance(ip_addrs, list), + True, + "List Public IP address should return valid IP address for network" + ) static_nat = ip_addrs[0] self.debug("Static NAT for VM: %s is: %s" % ( - self.virtual_machine.name, - static_nat.ipaddress - )) + self.virtual_machine.name, + static_nat.ipaddress + )) # Fetch details from user_ip_address table in database self.debug( - "select is_system from user_ip_address where public_ip_address='%s';" \ - % static_nat.ipaddress) + "select is_system from user_ip_address where\ + public_ip_address='%s';" % + static_nat.ipaddress) qresultset = self.dbclient.execute( - "select is_system from user_ip_address where public_ip_address='%s';" \ - % static_nat.ipaddress - ) + "select is_system from user_ip_address where\ + public_ip_address='%s';" % + static_nat.ipaddress) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "Check DB Query result set" - ) + len(qresultset), + 0, + "Check DB Query result set" + ) qresult = qresultset[0] self.assertEqual( - qresult[0], - 0, - "user_ip_address.is_system value should be 0" - ) + qresult[0], + 0, + "user_ip_address.is_system value should be 0" + ) - self.debug("Disassociate Static NAT: %s" % static_nat.ipaddress) + self.debug( + "Disassociate Static NAT: %s" % + static_nat.ipaddress) cmd = disassociateIpAddress.disassociateIpAddressCmd() cmd.id = static_nat.id self.apiclient.disassociateIpAddress(cmd) @@ -569,85 +583,86 @@ class TestEIP(cloudstackTestCase): # Fetch details from user_ip_address table in database self.debug( - "select state from user_ip_address where public_ip_address='%s';" \ - % static_nat.ipaddress) + "select state from user_ip_address where public_ip_address='%s';" + % static_nat.ipaddress) qresultset = self.dbclient.execute( - "select state from user_ip_address where public_ip_address='%s';" \ - % static_nat.ipaddress - ) + "select state from user_ip_address where public_ip_address='%s';" + % static_nat.ipaddress + ) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "Check DB Query result set" - ) + len(qresultset), + 0, + "Check DB Query result set" + ) qresult = qresultset[0] self.assertEqual( - qresult[0], - "Free", - "Ip should be marked as Free after disassociate IP" - ) + qresult[0], + "Free", + "Ip should be marked as Free after disassociate IP" + ) self.debug( "Fetching static NAT for VM: %s" % self.virtual_machine.name) ip_addrs = PublicIPAddress.list( - self.api_client, - associatednetworkid=self.guest_network.id, - isstaticnat=True, - account=self.account.name, - domainid=self.account.domainid, - listall=True - ) + self.api_client, + associatednetworkid=self.guest_network.id, + isstaticnat=True, + account=self.account.name, + domainid=self.account.domainid, + listall=True + ) self.assertEqual( - isinstance(ip_addrs, list), - True, - "List Public IP address should return valid IP address for network" - ) + isinstance(ip_addrs, list), + True, + "List Public IP address should return valid IP address for network" + ) static_nat = ip_addrs[0] self.debug("Static NAT for VM: %s is: %s" % ( - self.virtual_machine.name, - static_nat.ipaddress - )) + self.virtual_machine.name, + static_nat.ipaddress + )) # Fetch details from user_ip_address table in database self.debug( - "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \ - % static_nat.ipaddress) + "select is_system, one_to_one_nat from user_ip_address\ + where public_ip_address='%s';" % + static_nat.ipaddress) qresultset = self.dbclient.execute( - "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \ - % static_nat.ipaddress - ) + "select is_system, one_to_one_nat from user_ip_address\ + where public_ip_address='%s';" % + static_nat.ipaddress) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "Check DB Query result set" - ) + len(qresultset), + 0, + "Check DB Query result set" + ) qresult = qresultset[0] self.assertEqual( - qresult[0], - 1, - "is_system value should be 1 for automatically assigned IP" - ) + qresult[0], + 1, + "is_system value should be 1 for automatically assigned IP" + ) self.assertEqual( - qresult[1], - 1, - "one_to_one_nat value should be 1 for automatically assigned IP" - ) + qresult[1], + 1, + "one_to_one_nat value should be 1 for automatically assigned IP" + ) # try: # self.debug("SSH into VM: %s" % static_nat.ipaddress) # ssh = self.virtual_machine.get_ssh_client( @@ -657,48 +672,43 @@ class TestEIP(cloudstackTestCase): # (static_nat.ipaddress, e)) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) - try: - ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - 22, - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], - ) - self.debug("command: show ip") - res = ssh_client.execute("show ip") - result = str(res) - self.debug("Output: %s" % result) + self.services["netscaler"]["ipaddress"]) - self.assertEqual( - result.count(static_nat.ipaddress), - 1, - "One IP from EIP pool should be taken and configured on NS" - ) + ssh_client = SshClient( + self.services["netscaler"]["ipaddress"], + 22, + self.services["netscaler"]["username"], + self.services["netscaler"]["password"], + ) + self.debug("command: show ip") + res = ssh_client.execute("show ip") + result = str(res) + self.debug("Output: %s" % result) - self.debug("Command:show inat") - res = ssh_client.execute("show inat") + self.assertEqual( + result.count(static_nat.ipaddress), + 1, + "One IP from EIP pool should be taken and configured on NS" + ) - result = str(res) - self.debug("Output: %s" % result) + self.debug("Command:show inat") + res = ssh_client.execute("show inat") - self.assertEqual( - result.count("USIP: ON"), - 2, - "User source IP should be enabled for INAT service" - ) + result = str(res) + self.debug("Output: %s" % result) - except Exception as e: - self.fail("SSH Access failed for %s: %s" % \ - (self.services["netscaler"]["ipaddress"], e)) + self.assertEqual( + result.count("USIP: ON"), + 2, + "User source IP should be enabled for INAT service" + ) return - @attr(tags = ["eip"]) + @attr(tags=["eip"]) def test_04_disable_static_nat_system(self): """Test disable static NAT with system = True """ - # Validate the following # 1. Try to disassociate/disable static NAT on EIP where is_system=1 # 2. This operation should fail with proper error message. @@ -706,53 +716,56 @@ class TestEIP(cloudstackTestCase): self.debug( "Fetching static NAT for VM: %s" % self.virtual_machine.name) ip_addrs = PublicIPAddress.list( - self.api_client, - associatednetworkid=self.guest_network.id, - isstaticnat=True, - account=self.account.name, - domainid=self.account.domainid, - listall=True - ) + self.api_client, + associatednetworkid=self.guest_network.id, + isstaticnat=True, + account=self.account.name, + domainid=self.account.domainid, + listall=True + ) self.assertEqual( - isinstance(ip_addrs, list), - True, - "List Public IP address should return valid IP address for network" - ) + isinstance(ip_addrs, list), + True, + "List Public IP address should return valid IP address for network" + ) static_nat = ip_addrs[0] self.debug("Static NAT for VM: %s is: %s" % ( - self.virtual_machine.name, - static_nat.ipaddress - )) + self.virtual_machine.name, + static_nat.ipaddress + )) # Fetch details from user_ip_address table in database self.debug( - "select is_system from user_ip_address where public_ip_address='%s';" \ - % static_nat.ipaddress) + "select is_system from user_ip_address where\ + public_ip_address='%s';" % + static_nat.ipaddress) qresultset = self.dbclient.execute( - "select is_system from user_ip_address where public_ip_address='%s';" \ - % static_nat.ipaddress - ) + "select is_system from user_ip_address where\ + public_ip_address='%s';" % + static_nat.ipaddress) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "Check DB Query result set" - ) + len(qresultset), + 0, + "Check DB Query result set" + ) qresult = qresultset[0] self.assertEqual( - qresult[0], - 1, - "user_ip_address.is_system value should be 1" - ) + qresult[0], + 1, + "user_ip_address.is_system value should be 1" + ) - self.debug("Disassociate Static NAT: %s" % static_nat.ipaddress) + self.debug( + "Disassociate Static NAT: %s" % + static_nat.ipaddress) with self.assertRaises(Exception): cmd = disassociateIpAddress.disassociateIpAddressCmd() @@ -762,12 +775,11 @@ class TestEIP(cloudstackTestCase): self.debug("Disassociate system IP failed") return - @attr(tags = ["eip"]) + @attr(tags=["eip"]) def test_05_destroy_instance(self): """Test EIO after destroying instance """ - # Validate the following # 1. Destroy instance. Destroy should result in is_system=0 for EIP # and EIP should also be marked as free. @@ -778,104 +790,102 @@ class TestEIP(cloudstackTestCase): self.debug( "Fetching static NAT for VM: %s" % self.virtual_machine.name) ip_addrs = PublicIPAddress.list( - self.api_client, - associatednetworkid=self.guest_network.id, - isstaticnat=True, - account=self.account.name, - domainid=self.account.domainid, - listall=True - ) + self.api_client, + associatednetworkid=self.guest_network.id, + isstaticnat=True, + account=self.account.name, + domainid=self.account.domainid, + listall=True + ) self.assertEqual( - isinstance(ip_addrs, list), - True, - "List Public IP address should return valid IP address for network" - ) + isinstance(ip_addrs, list), + True, + "List Public IP address should return valid IP address for network" + ) static_nat = ip_addrs[0] self.debug("Static NAT for VM: %s is: %s" % ( - self.virtual_machine.name, - static_nat.ipaddress - )) + self.virtual_machine.name, + static_nat.ipaddress + )) self.debug("Destroying an instance: %s" % self.virtual_machine.name) self.virtual_machine.delete(self.apiclient, expunge=True) self.debug("Destroy instance complete!") vms = VirtualMachine.list( - self.apiclient, - id=self.virtual_machine.id - ) + self.apiclient, + id=self.virtual_machine.id + ) self.assertEqual( - vms, - None, - "list VM should not return anything after destroy" - ) + vms, + None, + "list VM should not return anything after destroy" + ) # Fetch details from user_ip_address table in database self.debug( - "select is_system, state from user_ip_address where public_ip_address='%s';" \ - % static_nat.ipaddress) + "select is_system, state from user_ip_address where\ + public_ip_address='%s';" % + static_nat.ipaddress) qresultset = self.dbclient.execute( - "select is_system, state from user_ip_address where public_ip_address='%s';" \ - % static_nat.ipaddress - ) + "select is_system, state from user_ip_address where\ + public_ip_address='%s';" % + static_nat.ipaddress) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "Check DB Query result set" - ) + len(qresultset), + 0, + "Check DB Query result set" + ) qresult = qresultset[0] self.assertEqual( - qresult[0], - 0, - "user_ip_address.is_system value should be 0" - ) + qresult[0], + 0, + "user_ip_address.is_system value should be 0" + ) self.assertEqual( - qresult[1], - "Free", - "IP should be marked as Free after destroying VM" - ) + qresult[1], + "Free", + "IP should be marked as Free after destroying VM" + ) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) - try: - ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - 22, - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], - ) - self.debug("command: show ip") - res = ssh_client.execute("show ip") - result = str(res) - self.debug("Output: %s" % result) + self.services["netscaler"]["ipaddress"]) - self.assertEqual( - result.count(static_nat.ipaddress), - 0, - "show ip should return nothing after VM destroy" - ) + ssh_client = SshClient( + self.services["netscaler"]["ipaddress"], + 22, + self.services["netscaler"]["username"], + self.services["netscaler"]["password"], + ) + self.debug("command: show ip") + res = ssh_client.execute("show ip") + result = str(res) + self.debug("Output: %s" % result) - self.debug("Command:show inat") - res = ssh_client.execute("show inat") + self.assertEqual( + result.count(static_nat.ipaddress), + 0, + "show ip should return nothing after VM destroy" + ) - result = str(res) - self.debug("Output: %s" % result) + self.debug("Command:show inat") + res = ssh_client.execute("show inat") - self.assertEqual( - result.count(static_nat.ipaddress), - 0, - "show inat should return nothing after VM destroy" - ) - except Exception as e: - self.fail("SSH Access failed for %s: %s" % \ - (self.services["netscaler"]["ipaddress"], e)) + result = str(res) + self.debug("Output: %s" % result) + + self.assertEqual( + result.count(static_nat.ipaddress), + 0, + "show inat should return nothing after VM destroy" + ) return @@ -891,70 +901,81 @@ class TestELB(cloudstackTestCase): cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.services['mode'] = cls.zone.networktype + + try: + cls.services["netscaler"] = cls.config.__dict__[ + "netscalerDevice"].__dict__ + except KeyError: + raise unittest.SkipTest("Please make sure you have included netscalerDevice\ + dict in your config file (keys - ipaddress, username,\ + password") + except Exception as e: + raise unittest.SkipTest(e) + cls.template = get_template( - cls.api_client, - cls.zone.id, - cls.services["ostype"] - ) + cls.api_client, + cls.zone.id, + cls.services["ostype"] + ) cls.services["virtual_machine"]["zoneid"] = cls.zone.id cls.services["virtual_machine"]["template"] = cls.template.id cls.service_offering = ServiceOffering.create( - cls.api_client, - cls.services["service_offering"] - ) + cls.api_client, + cls.services["service_offering"] + ) cls.account = Account.create( - cls.api_client, - cls.services["account"], - admin=True, - domainid=cls.domain.id - ) + cls.api_client, + cls.services["account"], + admin=True, + domainid=cls.domain.id + ) # Spawn an instance cls.vm_1 = VirtualMachine.create( - cls.api_client, - cls.services["virtual_machine"], - accountid=cls.account.name, - domainid=cls.account.domainid, - serviceofferingid=cls.service_offering.id - ) + cls.api_client, + cls.services["virtual_machine"], + accountid=cls.account.name, + domainid=cls.account.domainid, + serviceofferingid=cls.service_offering.id + ) cls.vm_2 = VirtualMachine.create( - cls.api_client, - cls.services["virtual_machine"], - accountid=cls.account.name, - domainid=cls.account.domainid, - serviceofferingid=cls.service_offering.id - ) + cls.api_client, + cls.services["virtual_machine"], + accountid=cls.account.name, + domainid=cls.account.domainid, + serviceofferingid=cls.service_offering.id + ) networks = Network.list( - cls.api_client, - zoneid=cls.zone.id, - listall=True - ) + cls.api_client, + zoneid=cls.zone.id, + listall=True + ) if isinstance(networks, list): # Basic zone has only one network i.e Basic network cls.guest_network = networks[0] else: raise Exception( "List networks returned empty response for zone: %s" % - cls.zone.id) + cls.zone.id) cls.lb_rule = LoadBalancerRule.create( - cls.api_client, - cls.services["lbrule"], - accountid=cls.account.name, - networkid=cls.guest_network.id, - domainid=cls.account.domainid - ) + cls.api_client, + cls.services["lbrule"], + accountid=cls.account.name, + networkid=cls.guest_network.id, + domainid=cls.account.domainid + ) cls.lb_rule.assign(cls.api_client, [cls.vm_1, cls.vm_2]) cls._cleanup = [ - cls.account, - cls.service_offering, - ] + cls.account, + cls.service_offering, + ] return @classmethod def tearDownClass(cls): try: - #Cleanup resources used + # Cleanup resources used cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) @@ -968,18 +989,17 @@ class TestELB(cloudstackTestCase): def tearDown(self): try: - #Clean up, terminate the created network offerings + # Clean up, terminate the created network offerings cleanup_resources(self.apiclient, self.cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return - - @attr(tags = ["eip"]) + + @attr(tags=["eip"]) def test_01_elb_create(self): """Test ELB by creating a LB rule """ - # Validate the following # 1. Deploy 2 instances # 2. Create LB rule to port 22 for the VMs and try to access VMs with @@ -994,22 +1014,22 @@ class TestELB(cloudstackTestCase): # Verify listSecurity groups response security_groups = SecurityGroup.list( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid + ) self.assertEqual( - isinstance(security_groups, list), - True, - "Check for list security groups response" - ) + isinstance(security_groups, list), + True, + "Check for list security groups response" + ) self.assertEqual( - len(security_groups), - 1, - "Check List Security groups response" - ) + len(security_groups), + 1, + "Check List Security groups response" + ) self.debug("List Security groups response: %s" % - str(security_groups)) + str(security_groups)) security_group = security_groups[0] @@ -1029,25 +1049,25 @@ class TestELB(cloudstackTestCase): self.debug( "Fetching LB IP for account: %s" % self.account.name) ip_addrs = PublicIPAddress.list( - self.api_client, - associatednetworkid=self.guest_network.id, - account=self.account.name, - domainid=self.account.domainid, - forloadbalancing=True, - listall=True - ) + self.api_client, + associatednetworkid=self.guest_network.id, + account=self.account.name, + domainid=self.account.domainid, + forloadbalancing=True, + listall=True + ) self.assertEqual( - isinstance(ip_addrs, list), - True, - "List Public IP address should return valid IP address for network" - ) + isinstance(ip_addrs, list), + True, + "List Public IP address should return valid IP address for network" + ) lb_ip = ip_addrs[0] self.debug("LB IP generated for account: %s is: %s" % ( - self.account.name, - lb_ip.ipaddress - )) -#TODO: uncomment this after ssh issue is resolved + self.account.name, + lb_ip.ipaddress + )) +# TODO: uncomment this after ssh issue is resolved # self.debug("SSHing into VMs using ELB IP: %s" % lb_ip.ipaddress) # try: # ssh_1 = self.vm_1.get_ssh_client(ipaddress=lb_ip.ipaddress) @@ -1086,50 +1106,51 @@ class TestELB(cloudstackTestCase): # Fetch details from user_ip_address table in database self.debug( - "select is_system from user_ip_address where public_ip_address='%s';" \ - % lb_ip.ipaddress) + "select is_system from user_ip_address where\ + public_ip_address='%s';" % + lb_ip.ipaddress) qresultset = self.dbclient.execute( - "select is_system from user_ip_address where public_ip_address='%s';" \ - % lb_ip.ipaddress - ) + "select is_system from user_ip_address where\ + public_ip_address='%s';" % + lb_ip.ipaddress) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "Check DB Query result set" - ) + len(qresultset), + 0, + "Check DB Query result set" + ) qresult = qresultset[0] self.assertEqual( - qresult[0], - 1, - "is_system value should be 1 for system generated LB rule" - ) + qresult[0], + 1, + "is_system value should be 1 for system generated LB rule" + ) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler"]["ipaddress"]) try: ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - 22, - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], - ) + self.services["netscaler"]["ipaddress"], + 22, + self.services["netscaler"]["username"], + self.services["netscaler"]["password"], + ) self.debug("command: show ip") res = ssh_client.execute("show ip") result = str(res) self.debug("Output: %s" % result) self.assertEqual( - result.count(lb_ip.ipaddress), - 1, - "One IP from EIP pool should be taken and configured on NS" - ) + result.count(lb_ip.ipaddress), + 1, + "One IP from EIP pool should be taken and configured on NS" + ) self.debug("Command:show lb vserver") res = ssh_client.execute("show lb vserver") @@ -1138,22 +1159,23 @@ class TestELB(cloudstackTestCase): self.debug("Output: %s" % result) self.assertEqual( - result.count("Cloud-VirtualServer-%s-22 (%s:22) - TCP" % (lb_ip.ipaddress, lb_ip.ipaddress)), - 1, - "User subnet IP should be enabled for LB service" - ) + result.count( + "Cloud-VirtualServer-%s-22 (%s:22) - TCP" % + (lb_ip.ipaddress, + lb_ip.ipaddress)), + 1, + "User subnet IP should be enabled for LB service") except Exception as e: - self.fail("SSH Access failed for %s: %s" % \ + self.fail("SSH Access failed for %s: %s" % (self.services["netscaler"]["ipaddress"], e)) return - - @attr(tags = ["eip"]) + + @attr(tags=["eip"]) def test_02_elb_acquire_and_create(self): """Test ELB by acquiring IP and then creating a LB rule """ - # Validate the following # 1. Deploy 2 instances # 2. Create LB rule to port 22 for the VMs and try to access VMs with @@ -1169,31 +1191,31 @@ class TestELB(cloudstackTestCase): self.debug("Acquiring new IP for network: %s" % self.guest_network.id) public_ip = PublicIPAddress.create( - self.apiclient, - accountid=self.account.name, - zoneid=self.zone.id, - domainid=self.account.domainid, - services=self.services["virtual_machine"] - ) + self.apiclient, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + services=self.services["virtual_machine"] + ) self.debug("IP address: %s is acquired by network: %s" % ( - public_ip.ipaddress, - self.guest_network.id)) + public_ip.ipaddress.ipaddress, + self.guest_network.id)) self.debug("Creating LB rule for public IP: %s" % - public_ip.ipaddress) + public_ip.ipaddress.ipaddress) lb_rule = LoadBalancerRule.create( - self.apiclient, - self.services["lbrule"], - accountid=self.account.name, - ipaddressid=public_ip.ipaddress.id, - networkid=self.guest_network.id, - domainid=self.account.domaind - ) + self.apiclient, + self.services["lbrule"], + accountid=self.account.name, + ipaddressid=public_ip.ipaddress.id, + networkid=self.guest_network.id, + domainid=self.account.domainid + ) self.debug("Assigning VMs (%s, %s) to LB rule: %s" % (self.vm_1.name, self.vm_2.name, lb_rule.name)) lb_rule.assign(self.apiclient, [self.vm_1, self.vm_2]) -#TODO: workaround : add route in the guest VM for SNIP +# TODO: workaround : add route in the guest VM for SNIP # # self.debug("SSHing into VMs using ELB IP: %s" % # public_ip.ipaddress) @@ -1233,52 +1255,54 @@ class TestELB(cloudstackTestCase): # self.fail( # "SSH Access failed for %s: %s" % (self.vm_1.ipaddress, e)) # -## Fetch details from user_ip_address table in database +# Fetch details from user_ip_address table in database self.debug( - "select is_system from user_ip_address where public_ip_address='%s';" \ - % public_ip.ipaddress) + "select is_system from user_ip_address where\ + public_ip_address='%s';" % + public_ip.ipaddress.ipaddress) qresultset = self.dbclient.execute( - "select is_system from user_ip_address where public_ip_address='%s';" \ - % public_ip.ipaddress) + "select is_system from user_ip_address where\ + public_ip_address='%s';" % + public_ip.ipaddress.ipaddress) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "Check DB Query result set" - ) + len(qresultset), + 0, + "Check DB Query result set" + ) qresult = qresultset[0] self.assertEqual( - qresult[0], - 0, - "is_system value should be 0 for non-system generated LB rule" - ) + qresult[0], + 0, + "is_system value should be 0 for non-system generated LB rule" + ) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler"]["ipaddress"]) try: ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - 22, - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], - ) + self.services["netscaler"]["ipaddress"], + 22, + self.services["netscaler"]["username"], + self.services["netscaler"]["password"], + ) self.debug("command: show ip") res = ssh_client.execute("show ip") result = str(res) self.debug("Output: %s" % result) self.assertEqual( - result.count(public_ip.ipaddress), - 1, - "One IP from EIP pool should be taken and configured on NS" - ) + result.count(public_ip.ipaddress.ipaddress), + 1, + "One IP from EIP pool should be taken and configured on NS" + ) self.debug("Command:show lb vserver") res = ssh_client.execute("show lb vserver") @@ -1287,23 +1311,23 @@ class TestELB(cloudstackTestCase): self.debug("Output: %s" % result) self.assertEqual( - result.count("Cloud-VirtualServer-%s-22 (%s:22) - TCP" % (public_ip.ipaddress, public_ip.ipaddress)), - 1, - "User subnet IP should be enabled for LB service" - ) + result.count( + "Cloud-VirtualServer-%s-22 (%s:22) - TCP" % + (public_ip.ipaddress.ipaddress, + public_ip.ipaddress.ipaddress)), + 1, + "User subnet IP should be enabled for LB service") except Exception as e: - self.fail("SSH Access failed for %s: %s" % \ + self.fail("SSH Access failed for %s: %s" % (self.services["netscaler"]["ipaddress"], e)) return - - @attr(tags = ["eip"]) + @attr(tags=["eip"]) def test_03_elb_delete_lb_system(self): """Test delete LB rule generated with public IP with is_system = 1 """ - # Validate the following # 1. Deleting LB rule should release EIP where is_system=1 # 2. check configuration changes for EIP reflects on NS @@ -1315,73 +1339,69 @@ class TestELB(cloudstackTestCase): self.debug( "Fetching LB IP for account: %s" % self.account.name) ip_addrs = PublicIPAddress.list( - self.api_client, - associatednetworkid=self.guest_network.id, - account=self.account.name, - domainid=self.account.domainid, - forloadbalancing=True, - listall=True - ) + self.api_client, + associatednetworkid=self.guest_network.id, + account=self.account.name, + domainid=self.account.domainid, + forloadbalancing=True, + listall=True + ) self.assertEqual( - isinstance(ip_addrs, list), - True, - "List Public IP address should return valid IP address for network" - ) + isinstance(ip_addrs, list), + True, + "List Public IP address should return valid IP address for network" + ) lb_ip = ip_addrs[0] self.debug("LB IP generated for account: %s is: %s" % ( - self.account.name, - lb_ip.ipaddress - )) + self.account.name, + lb_ip.ipaddress + )) self.debug("Deleting LB rule: %s" % self.lb_rule.id) self.lb_rule.delete(self.apiclient) - time.sleep(60) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) - try: - ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - 22, - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], - ) - self.debug("command: show ip") - res = ssh_client.execute("show ip") - result = str(res) - self.debug("Output: %s" % result) + self.services["netscaler"]["ipaddress"]) - self.assertEqual( - result.count(lb_ip.ipaddress), - 0, - "One IP from EIP pool should be taken and configured on NS" - ) + ssh_client = SshClient( + self.services["netscaler"]["ipaddress"], + 22, + self.services["netscaler"]["username"], + self.services["netscaler"]["password"], + ) + self.debug("command: show ip") + res = ssh_client.execute("show ip") + result = str(res) + self.debug("Output: %s" % result) - self.debug("Command:show lb vserver") - res = ssh_client.execute("show lb vserver") + self.assertEqual( + result.count(lb_ip.ipaddress), + 1, + "One IP from EIP pool should be taken and configured on NS" + ) - result = str(res) - self.debug("Output: %s" % result) - - self.assertEqual( - result.count("Cloud-VirtualServer-%s-22 (%s:22) - TCP" % (lb_ip.ipaddress, lb_ip.ipaddress) ), - 0, - "User subnet IP should be enabled for LB service" - ) + self.debug("Command:show lb vserver") + res = ssh_client.execute("show lb vserver") - except Exception as e: - self.fail("SSH Access failed for %s: %s" % \ - (self.services["netscaler"]["ipaddress"], e)) + result = str(res) + self.debug("Output: %s" % result) + + self.assertEqual( + result.count( + "Cloud-VirtualServer-%s-22 (%s:22) - TCP" % + (lb_ip.ipaddress, + lb_ip.ipaddress)), + 1, + "User subnet IP should be enabled for LB service") return - - @attr(tags = ["eip"]) + + @attr(tags=["eip"]) def test_04_delete_lb_on_eip(self): """Test delete LB rule generated on EIP """ - # Validate the following # 1. Deleting LB rule won't release EIP where is_system=0 # 2. disassociateIP must release the above IP @@ -1391,77 +1411,79 @@ class TestELB(cloudstackTestCase): # * "show lb vserer"-make sure that output says they are all up and # running and USNIP : ON - # Fetch details from account_id table in database + # Fetch details from account_id table in database self.debug( - "select id from account where account_name='%s';" \ - % self.account.name) + "select id from account where account_name='%s';" + % self.account.name) qresultset = self.dbclient.execute( - "select id from account where account_name='%s';" \ - % self.account.name) + "select id from account where account_name='%s';" + % self.account.name) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "DB query should return a valid public IP address" - ) + len(qresultset), + 0, + "DB query should return a valid public IP address" + ) qresult = qresultset[0] account_id = qresult[0] # Fetch details from user_ip_address table in database self.debug( - "select public_ip_address from user_ip_address where is_system=0 and account_id=%s;" \ - % account_id) + "select public_ip_address from user_ip_address where\ + is_system=0 and account_id=%s;" % + account_id) qresultset = self.dbclient.execute( - "select public_ip_address from user_ip_address where is_system=0 and account_id=%s;" \ - % account_id) + "select public_ip_address from user_ip_address where\ + is_system=0 and account_id=%s;" % + account_id) self.assertEqual( - isinstance(qresultset, list), - True, - "Check DB query result set for valid data" - ) + isinstance(qresultset, list), + True, + "Check DB query result set for valid data" + ) self.assertNotEqual( - len(qresultset), - 0, - "DB query should return a valid public IP address" - ) + len(qresultset), + 0, + "DB query should return a valid public IP address" + ) qresult = qresultset[0] public_ip = qresult[0] self.debug( "Fetching public IP for account: %s" % self.account.name) ip_addrs = PublicIPAddress.list( - self.api_client, - ipaddress=public_ip, - listall=True - ) + self.api_client, + ipaddress=public_ip, + listall=True + ) self.debug("ip address list: %s" % ip_addrs) self.assertEqual( - isinstance(ip_addrs, list), - True, - "List Public IP address should return valid IP address for network" - ) + isinstance(ip_addrs, list), + True, + "List Public IP address should return valid IP address for network" + ) lb_ip = ip_addrs[0] lb_rules = LoadBalancerRule.list( - self.apiclient, - publicipid=lb_ip.id, - listall=True - ) + self.apiclient, + publicipid=lb_ip.id, + listall=True + ) self.assertEqual( - isinstance(lb_rules, list), - True, - "Atleast one LB rule must be present for public IP address" - ) + isinstance(lb_rules, list), + True, + "Atleast one LB rule must be present for public IP address" + ) lb_rule = lb_rules[0] self.debug("Deleting LB rule associated with IP: %s" % public_ip) @@ -1472,38 +1494,38 @@ class TestELB(cloudstackTestCase): except Exception as e: self.fail("Deleting LB rule failed for IP: %s-%s" % (public_ip, e)) -#TODO:check the lb rule list and then confirm that lb rule is deleted +# TODO:check the lb rule list and then confirm that lb rule is deleted self.debug("LB rule deleted!") ip_addrs = PublicIPAddress.list( - self.api_client, - ipaddress=public_ip, - listall=True - ) + self.api_client, + ipaddress=public_ip, + listall=True + ) self.assertEqual( - isinstance(ip_addrs, list), - True, - "Deleting LB rule should not delete public IP" - ) + isinstance(ip_addrs, list), + True, + "Deleting LB rule should not delete public IP" + ) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler"]["ipaddress"]) try: ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - 22, - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], - ) + self.services["netscaler"]["ipaddress"], + 22, + self.services["netscaler"]["username"], + self.services["netscaler"]["password"], + ) self.debug("command: show ip") res = ssh_client.execute("show ip") result = str(res) self.debug("Output: %s" % result) self.assertNotEqual( - result.count(public_ip), - 1, - "One IP from EIP pool should be taken and configured on NS" - ) + result.count(public_ip), + 1, + "One IP from EIP pool should be taken and configured on NS" + ) self.debug("Command:show lb vserver") res = ssh_client.execute("show lb vserver") @@ -1512,12 +1534,14 @@ class TestELB(cloudstackTestCase): self.debug("Output: %s" % result) self.assertNotEqual( - result.count("Cloud-VirtualServer-%s-22 (%s:22) - TCP" % (lb_ip.ipaddress, lb_ip.ipaddress)), - 1, - "User subnet IP should be enabled for LB service" - ) + result.count( + "Cloud-VirtualServer-%s-22 (%s:22) - TCP" % + (lb_ip.ipaddress, + lb_ip.ipaddress)), + 1, + "User subnet IP should be enabled for LB service") except Exception as e: - self.fail("SSH Access failed for %s: %s" % \ + self.fail("SSH Access failed for %s: %s" % (self.services["netscaler"]["ipaddress"], e)) return