http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f583dcb4/test/integration/component/test_persistent_networks.py ---------------------------------------------------------------------- diff --git a/test/integration/component/test_persistent_networks.py b/test/integration/component/test_persistent_networks.py index 02f6229..65b109f 100644 --- a/test/integration/component/test_persistent_networks.py +++ b/test/integration/component/test_persistent_networks.py @@ -48,8 +48,10 @@ from marvin.cloudstackTestCase import cloudstackTestCase, unittest from ddt import ddt, data import time + @ddt class TestPersistentNetworks(cloudstackTestCase): + ''' Test Persistent Networks without running VMs ''' @@ -65,44 +67,59 @@ class TestPersistentNetworks(cloudstackTestCase): cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.template = get_template( - cls.api_client, - cls.zone.id, - cls.services["ostype"] - ) + cls.api_client, + cls.zone.id, + cls.services["ostype"] + ) cls.account = Account.create( - cls.api_client, - cls.services["account"], - domainid=cls.domain.id - ) + cls.api_client, + cls.services["account"], + domainid=cls.domain.id + ) 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.isolated_persistent_network_offering = cls.createNetworkOffering("nw_off_isolated_persistent") - cls.isolated_persistent_network_offering_netscaler = cls.createNetworkOffering("nw_off_isolated_persistent_netscaler") - cls.isolated_persistent_network_offering_RVR = cls.createNetworkOffering("nw_off_persistent_RVR") - cls.isolated_network_offering = cls.createNetworkOffering("isolated_network_offering") - cls.isolated_network_offering_netscaler = cls.createNetworkOffering("nw_off_isolated_netscaler") + cls.api_client, + cls.services["service_offering"] + ) + cls.isolated_persistent_network_offering = cls.createNetworkOffering( + "nw_off_isolated_persistent") + cls.isolated_persistent_network_offering_netscaler =\ + cls.createNetworkOffering( + "nw_off_isolated_persistent_netscaler" + ) + cls.isolated_persistent_network_offering_RVR =\ + cls.createNetworkOffering( + "nw_off_persistent_RVR" + ) + cls.isolated_network_offering = cls.createNetworkOffering( + "isolated_network_offering") + cls.isolated_network_offering_netscaler = cls.createNetworkOffering( + "nw_off_isolated_netscaler") # Configure Netscaler device - # If configuration succeeds, set ns_configured to True so that Netscaler tests are executed + # If configuration succeeds, set ns_configured to True so that + # Netscaler tests are executed cls.ns_configured = False try: - cls.netscaler = add_netscaler(cls.api_client, cls.zone.id, cls.services["netscaler_VPX"]) + cls.netscaler = add_netscaler( + cls.api_client, + cls.zone.id, + cls.services["netscaler_VPX"]) cls._cleanup.append(cls.netscaler) cls.ns_configured = True except Exception: cls.ns_configured = False - # network will be deleted as part of account cleanup cls._cleanup = [ - cls.account, cls.service_offering, cls.isolated_persistent_network_offering, cls.isolated_network_offering, - cls.isolated_persistent_network_offering_RVR, cls.isolated_persistent_network_offering_netscaler, - cls.isolated_network_offering_netscaler - ] + cls.account, + cls.service_offering, + cls.isolated_persistent_network_offering, + cls.isolated_network_offering, + cls.isolated_persistent_network_offering_RVR, + cls.isolated_persistent_network_offering_netscaler, + cls.isolated_network_offering_netscaler] return @classmethod @@ -117,13 +134,16 @@ class TestPersistentNetworks(cloudstackTestCase): @classmethod def createNetworkOffering(cls, network_offering_type): network_offering = NetworkOffering.create( - cls.api_client, - cls.services[network_offering_type], - conservemode=False - ) + cls.api_client, + cls.services[network_offering_type], + conservemode=False + ) # Update network offering state from disabled to enabled. - NetworkOffering.update(network_offering, cls.api_client, id=network_offering.id, - state="enabled") + NetworkOffering.update( + network_offering, + cls.api_client, + id=network_offering.id, + state="enabled") return network_offering def checkRouterAccessibility(self, router): @@ -132,32 +152,41 @@ class TestPersistentNetworks(cloudstackTestCase): hypervisor = str(get_hypervisor_type(self.api_client)) if hypervisor.lower() in ('vmware', 'hyperv'): - #SSH is done via management server for Vmware and Hyper-V + # SSH is done via management server for Vmware and Hyper-V sourceip = self.api_client.connection.mgtSvr else: - #For others, we will have to get the ipaddress of host connected to vm - hosts = Host.list(self.api_client,id=router.hostid) - self.assertEqual(validateList(hosts)[0], PASS, "hosts list validation failed, list is %s" % hosts) + # For others, we will have to get the ipaddress of host connected + # to vm + hosts = Host.list(self.api_client, id=router.hostid) + self.assertEqual( + validateList(hosts)[0], + PASS, + "hosts list validation failed, list is %s" % + hosts) host = hosts[0] sourceip = host.ipaddress # end else try: - sshClient = SshClient(host=sourceip, port=22, user='root',passwd=self.services["host_password"]) + sshClient = SshClient( + host=sourceip, + port=22, + user='root', + passwd=self.services["host_password"]) res = sshClient.execute("ping -c 1 %s" % ( - router.linklocalip - )) + router.linklocalip + )) self.debug("SSH result: %s" % res) except Exception as e: - self.fail("SSH Access failed for %s: %s" % \ + self.fail("SSH Access failed for %s: %s" % (sourceip, e) ) result = str(res) self.assertEqual( - result.count("1 received"), - 1, - "Ping to router should be successful" - ) + result.count("1 received"), + 1, + "Ping to router should be successful" + ) return def verifyVmExpungement(self, virtual_machine): @@ -184,7 +213,7 @@ class TestPersistentNetworks(cloudstackTestCase): def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() - self.cleanup = [ ] + self.cleanup = [] return def tearDown(self): @@ -204,29 +233,44 @@ class TestPersistentNetworks(cloudstackTestCase): # 3. destroy created virtual machine # # validation - # 1. Persistent network state should be implemented before VM creation and have some vlan assigned + # 1. Persistent network state should be implemented before VM creation + # and have some vlan assigned # 2. virtual machine should be created successfully - # 3. Network state should be implemented even after destroying all vms in network + # 3. Network state should be implemented even after destroying all vms + # in network # Creating isolated persistent network - network = Network.create(self.apiclient,self.services["isolated_network"], - networkofferingid=self.isolated_persistent_network_offering.id, - accountid=self.account.name,domainid=self.domain.id, - zoneid=self.zone.id) + network = Network.create( + self.apiclient, + self.services["isolated_network"], + networkofferingid=self.isolated_persistent_network_offering.id, + accountid=self.account.name, + domainid=self.domain.id, + zoneid=self.zone.id) self.cleanup.append(network) - response = verifyNetworkState(self.apiclient, network.id, "implemented") + response = verifyNetworkState( + self.apiclient, + network.id, + "implemented") exceptionOccured = response[0] isNetworkInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccured or (not isNetworkInDesiredState)): - self.fail(exceptionMessage) - self.assertIsNotNone(network.vlan, "vlan must not be null for persistent network") + self.fail(exceptionMessage) + self.assertIsNotNone( + network.vlan, + "vlan must not be null for persistent network") try: - virtual_machine = VirtualMachine.create(self.apiclient,self.services["virtual_machine"], - networkids=[network.id],serviceofferingid=self.service_offering.id, - accountid=self.account.name,domainid=self.domain.id) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.domain.id) virtual_machine.delete(self.apiclient) except Exception as e: self.fail("vm creation failed: %s" % e) @@ -235,8 +279,11 @@ class TestPersistentNetworks(cloudstackTestCase): self.verifyVmExpungement(virtual_machine) # wait for time such that, network is cleaned up - # assuming that it will change its state to allocated after this much period - wait_for_cleanup(self.api_client, ["network.gc.interval", "network.gc.wait"]) + # assuming that it will change its state to allocated after this much + # period + wait_for_cleanup( + self.api_client, [ + "network.gc.interval", "network.gc.wait"]) verifyNetworkState(self.api_client, network.id, "implemented") exceptionOccured = response[0] @@ -244,7 +291,7 @@ class TestPersistentNetworks(cloudstackTestCase): exceptionMessage = response[2] if (exceptionOccured or (not isNetworkInDesiredState)): - self.fail(exceptionMessage) + self.fail(exceptionMessage) return @attr(tags=["advanced"], required_hardware="false") @@ -255,24 +302,32 @@ class TestPersistentNetworks(cloudstackTestCase): # validation # 1. network offering should throw an exception try: - shared_persistent_network_offering = self.createNetworkOffering("nw_offering_shared_persistent") + shared_persistent_network_offering = self.createNetworkOffering( + "nw_offering_shared_persistent") shared_persistent_network_offering.delete(self.apiclient) - self.fail("Shared network got created with ispersistent flag True in the offering, it should have failed") + self.fail( + "Shared network got created with ispersistent flag\ + True in the offering, it should have failed") except Exception: pass @attr(tags=["advanced"]) def test_persistent_network_offering_with_VPCVR_services(self): # steps - # 1. create network offering with persistent field enabled and all the services through VpcVirtualRouter + # 1. create network offering with persistent field enabled and + # all the services through VpcVirtualRouter # # validation # 1. network offering should be created successfully try: - persistent_network_offering_VPCVR = self.createNetworkOffering("nw_off_persistent_VPCVR_LB") + persistent_network_offering_VPCVR = self.createNetworkOffering( + "nw_off_persistent_VPCVR_LB") persistent_network_offering_VPCVR.delete(self.apiclient) except Exception as e: - self.fail("Failed creating persistent network offering with VPCVR services: %s" % e) + self.fail( + "Failed creating persistent network offering\ + with VPCVR services: %s" % + e) @attr(tags=["advanced"]) def test_list_persistent_network_offering(self): @@ -282,24 +337,36 @@ class TestPersistentNetworks(cloudstackTestCase): # # validation # 1. ispersistent flag should list as True - network_offering = self.createNetworkOffering("nw_off_isolated_persistent") + network_offering = self.createNetworkOffering( + "nw_off_isolated_persistent") self.cleanup.append(network_offering) - nw_offering_list = NetworkOffering.list(self.api_client, id=network_offering.id) - self.assertEqual(validateList(nw_offering_list)[0], PASS, "network offerings' list validation failed, list is %s" % - nw_offering_list) - self.assertEqual(nw_offering_list[0].ispersistent, True, "ispersistent flag should be true for the network offering") + nw_offering_list = NetworkOffering.list( + self.api_client, + id=network_offering.id) + self.assertEqual( + validateList(nw_offering_list)[0], + PASS, + "network offerings' list validation failed, list is %s" % + nw_offering_list) + self.assertEqual( + nw_offering_list[0].ispersistent, + True, + "ispersistent flag should be true for the network offering") return - @data("LB-VR","LB-NS") - @attr(tags=["advanced","advancedns"]) + @data("LB-VR", "LB-NS") + @attr(tags=["advanced", "advancedns"]) def test_upgrade_to_persistent_services_VR(self, value): - # This test is run against two networks (one with LB as virtual router and other one is LB with Netscaler) + # This test is run against two networks (one with LB as virtual router + # and other one is LB with Netscaler) # All other services through VR # steps - # 1. create isolated network with network offering which has ispersistent field disabled - # 2. upgrade isolated network offering to network offering which has ispersistent field enabled + # 1. create isolated network with network offering which + # has ispersistent field disabled + # 2. upgrade isolated network offering to network offering + # which has ispersistent field enabled # 3. Deploy VM ,acquire IP, create Firewall, NAT rules # 4. Verify the working of NAT, Firewall rules # 5. Delete VM @@ -308,12 +375,16 @@ class TestPersistentNetworks(cloudstackTestCase): # validation # 1. update of network should happen successfully # 2. NAT and Firewall rule should work as expected - # 3. After network clenaup interval, network state should be implemented and have some vlan assigned + # 3. After network clenaup interval, network state should be + # implemented and have some vlan assigned - # Set network offering as required (either LB through VR or LB through Netscaler) + # Set network offering as required (either LB through VR or LB through + # Netscaler) networkOffering = self.isolated_network_offering - changecidr = False # This will be true in case of Netscaler case, you have to change cidr while updating network + # This will be true in case of Netscaler case, you have to change cidr + # while updating network + changecidr = False # In case Netscaler is used for LB if value == "LB-NS": @@ -321,56 +392,90 @@ class TestPersistentNetworks(cloudstackTestCase): networkOffering = self.isolated_network_offering_netscaler changecidr = True else: - self.skipTest("Skipping - this test required netscaler configured in the network") + self.skipTest( + "Skipping - this test required netscaler\ + configured in the network") # Create Account - account = Account.create(self.api_client, self.services["account"],domainid=self.domain.id) + account = Account.create( + self.api_client, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(account) # create network with the appropriate offering (not persistent) - isolated_network = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=networkOffering.id, - accountid=account.name,domainid=self.domain.id, - zoneid=self.zone.id) - response = verifyNetworkState(self.api_client, isolated_network.id, "allocated") + isolated_network = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=networkOffering.id, + accountid=account.name, + domainid=self.domain.id, + zoneid=self.zone.id) + response = verifyNetworkState( + self.api_client, + isolated_network.id, + "allocated") exceptionOccured = response[0] isNetworkInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccured or (not isNetworkInDesiredState)): - self.fail(exceptionMessage) + self.fail(exceptionMessage) # Update the network with persistent network offering - isolated_network.update(self.apiclient, networkofferingid=self.isolated_persistent_network_offering.id, changecidr=changecidr) + isolated_network.update( + self.apiclient, + networkofferingid=self.isolated_persistent_network_offering.id, + changecidr=changecidr) try: - virtual_machine = VirtualMachine.create(self.apiclient,self.services["virtual_machine"], - networkids=[isolated_network.id], - serviceofferingid=self.service_offering.id, - accountid=account.name,domainid=self.domain.id) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + isolated_network.id], + serviceofferingid=self.service_offering.id, + accountid=account.name, + domainid=self.domain.id) except Exception as e: self.fail("vm creation failed: %s" % e) # Acquire public ip and open firewall for it - self.debug("Associating public IP for network: %s" % isolated_network.id) - ipaddress = PublicIPAddress.create(self.api_client,accountid=account.name, - zoneid=self.zone.id,domainid=account.domainid, - networkid=isolated_network.id) - - FireWallRule.create(self.apiclient,ipaddressid=ipaddress.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) + self.debug( + "Associating public IP for network: %s" % + isolated_network.id) + ipaddress = PublicIPAddress.create( + self.api_client, + accountid=account.name, + zoneid=self.zone.id, + domainid=account.domainid, + networkid=isolated_network.id) + + FireWallRule.create( + self.apiclient, + ipaddressid=ipaddress.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) # Create NAT rule - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=ipaddress.ipaddress.id, - networkid=isolated_network.id) + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=ipaddress.ipaddress.id, + networkid=isolated_network.id) # Check if SSH works try: - virtual_machine.get_ssh_client(ipaddress=ipaddress.ipaddress.ipaddress) + virtual_machine.get_ssh_client( + ipaddress=ipaddress.ipaddress.ipaddress) except Exception as e: - self.fail("Exception while SSHing to VM %s with IP %s" % (virtual_machine.id, ipaddress.ipaddress.ipaddress)) + self.fail( + "Exception while SSHing to VM %s with IP %s" % + (virtual_machine.id, ipaddress.ipaddress.ipaddress)) # Delete VM virtual_machine.delete(self.api_client) @@ -379,23 +484,31 @@ class TestPersistentNetworks(cloudstackTestCase): self.verifyVmExpungement(virtual_machine) # wait for time such that, network is cleaned up - wait_for_cleanup(self.api_client, ["network.gc.interval", "network.gc.wait"]) - - # Check network state now, this will bolster that network updation has taken effect - response = verifyNetworkState(self.api_client, isolated_network.id, "implemented") + wait_for_cleanup( + self.api_client, [ + "network.gc.interval", "network.gc.wait"]) + + # Check network state now, this will bolster that network updation has + # taken effect + response = verifyNetworkState( + self.api_client, + isolated_network.id, + "implemented") exceptionOccured = response[0] isNetworkInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccured or (not isNetworkInDesiredState)): - self.fail(exceptionMessage) + self.fail(exceptionMessage) return @attr(tags=["advanced"]) def test_upgrade_network_VR_to_PersistentRVR(self): # steps - # 1. create isolated network with network offering which has ispersistent field disabled (services through VR) - # 2. upgrade isolated network offering to network offering which has ispersistent field enabled (with RVR) + # 1. create isolated network with network offering which has + # ispersistent field disabled (services through VR) + # 2. upgrade isolated network offering to network offering which + # has ispersistent field enabled (with RVR) # 3. Deploy VM ,acquire IP, create Firewall, NAT rules # 4. Verify the working of NAT, Firewall rules # 5. Delete VM @@ -404,53 +517,85 @@ class TestPersistentNetworks(cloudstackTestCase): # validation # 1. update of network should happen successfully # 2. NAT and Firewall rule should work as expected - # 3. After network clenaup interval, network state should be implemented and have some vlan assigned + # 3. After network clenaup interval, network state should be + # implemented and have some vlan assigned # Create Account and isolated network in it - account = Account.create(self.api_client, self.services["account"],domainid=self.domain.id) + account = Account.create( + self.api_client, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(account) - isolated_network = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=self.isolated_network_offering.id, - accountid=account.name,domainid=self.domain.id, - zoneid=self.zone.id) - response = verifyNetworkState(self.api_client, isolated_network.id, "allocated") + isolated_network = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=self.isolated_network_offering.id, + accountid=account.name, + domainid=self.domain.id, + zoneid=self.zone.id) + response = verifyNetworkState( + self.api_client, + isolated_network.id, + "allocated") exceptionOccured = response[0] isNetworkInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccured or (not isNetworkInDesiredState)): - self.fail(exceptionMessage) + self.fail(exceptionMessage) # Update network with network offering which has RVR - isolated_network.update(self.apiclient, networkofferingid=self.isolated_persistent_network_offering_RVR.id) + isolated_network.update( + self.apiclient, + networkofferingid=self.isolated_persistent_network_offering_RVR.id) # Deploy VM try: - virtual_machine = VirtualMachine.create(self.apiclient,self.services["virtual_machine"], - networkids=[isolated_network.id], - serviceofferingid=self.service_offering.id, - accountid=account.name,domainid=self.domain.id) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + isolated_network.id], + serviceofferingid=self.service_offering.id, + accountid=account.name, + domainid=self.domain.id) except Exception as e: self.fail("vm creation failed: %s" % e) # Acquire public IP and open firewall rule, create NAT rule - self.debug("Associating public IP for network: %s" % isolated_network.id) - ipaddress = PublicIPAddress.create(self.api_client,accountid=account.name, - zoneid=self.zone.id,domainid=account.domainid, - networkid=isolated_network.id) - - FireWallRule.create(self.apiclient,ipaddressid=ipaddress.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=ipaddress.ipaddress.id, - networkid=isolated_network.id) + self.debug( + "Associating public IP for network: %s" % + isolated_network.id) + ipaddress = PublicIPAddress.create( + self.api_client, + accountid=account.name, + zoneid=self.zone.id, + domainid=account.domainid, + networkid=isolated_network.id) + + FireWallRule.create( + self.apiclient, + ipaddressid=ipaddress.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=ipaddress.ipaddress.id, + networkid=isolated_network.id) try: - virtual_machine.get_ssh_client(ipaddress=ipaddress.ipaddress.ipaddress) + virtual_machine.get_ssh_client( + ipaddress=ipaddress.ipaddress.ipaddress) except Exception as e: - self.fail("Exception while SSHing to VM %s with IP %s" % (virtual_machine.id, ipaddress.ipaddress.ipaddress)) + self.fail( + "Exception while SSHing to VM %s with IP %s" % + (virtual_machine.id, ipaddress.ipaddress.ipaddress)) virtual_machine.delete(self.api_client) @@ -458,24 +603,32 @@ class TestPersistentNetworks(cloudstackTestCase): self.verifyVmExpungement(virtual_machine) # wait for time such that, network is cleaned up - wait_for_cleanup(self.api_client, ["network.gc.interval", "network.gc.wait"]) - - # Check network state now, this will bolster that network updation has taken effect - response = verifyNetworkState(self.api_client, isolated_network.id, "implemented") + wait_for_cleanup( + self.api_client, [ + "network.gc.interval", "network.gc.wait"]) + + # Check network state now, this will bolster that network updation has + # taken effect + response = verifyNetworkState( + self.api_client, + isolated_network.id, + "implemented") exceptionOccured = response[0] isNetworkInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccured or (not isNetworkInDesiredState)): - self.fail(exceptionMessage) + self.fail(exceptionMessage) return @attr(tags=["advanced", "advancedns"]) def test_upgrade_network_NS_to_persistent_NS(self): # steps - # 1. create isolated network with network offering which has ispersistent field disabled + # 1. create isolated network with network offering which + # has ispersistent field disabled # and LB service through Netscaler - # 2. upgrade isolated network offering to network offering which has ispersistent field enabled + # 2. upgrade isolated network offering to network offering + # which has ispersistent field enabled # and LB service through Netscaler # 3. Deploy VM ,acquire IP, create Firewall, NAT rules # 4. Verify the working of NAT, Firewall rules @@ -485,55 +638,91 @@ class TestPersistentNetworks(cloudstackTestCase): # validation # 1. update of network should happen successfully # 2. NAT and Firewall rule should work as expected - # 3. After network clenaup interval, network state should be implemented and have some vlan assigned + # 3. After network clenaup interval, network state should be + # implemented and have some vlan assigned # Skip test if Netscaler is not configured if not self.ns_configured: - self.skipTest("Skipping - this test required netscaler configured in the network") + self.skipTest( + "Skipping - this test required netscaler\ + configured in the network") # Create Account and create isolated network in it - account = Account.create(self.api_client, self.services["account"],domainid=self.domain.id) + account = Account.create( + self.api_client, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(account) - isolated_network = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=self.isolated_network_offering_netscaler.id, - accountid=account.name,domainid=self.domain.id, - zoneid=self.zone.id) - response = verifyNetworkState(self.api_client, isolated_network.id, "allocated") + isolated_network = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=self.isolated_network_offering_netscaler.id, + accountid=account.name, + domainid=self.domain.id, + zoneid=self.zone.id) + response = verifyNetworkState( + self.api_client, + isolated_network.id, + "allocated") exceptionOccured = response[0] isNetworkInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccured or (not isNetworkInDesiredState)): - self.fail(exceptionMessage) + self.fail(exceptionMessage) - isolated_network.update(self.apiclient, networkofferingid=self.isolated_persistent_network_offering_netscaler.id, changecidr=True) + isolated_network.update( + self.apiclient, + networkofferingid=( + self.isolated_persistent_network_offering_netscaler.id), + changecidr=True) # Deploy VM try: - virtual_machine = VirtualMachine.create(self.apiclient,self.services["virtual_machine"], - networkids=[isolated_network.id], - serviceofferingid=self.service_offering.id, - accountid=account.name,domainid=self.domain.id) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + isolated_network.id], + serviceofferingid=self.service_offering.id, + accountid=account.name, + domainid=self.domain.id) except Exception as e: self.fail("vm creation failed: %s" % e) - self.debug("Associating public IP for network: %s" % isolated_network.id) - ipaddress = PublicIPAddress.create(self.api_client,accountid=account.name, - zoneid=self.zone.id,domainid=account.domainid, - networkid=isolated_network.id) - - FireWallRule.create(self.apiclient,ipaddressid=ipaddress.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=ipaddress.ipaddress.id, - networkid=isolated_network.id) + self.debug( + "Associating public IP for network: %s" % + isolated_network.id) + ipaddress = PublicIPAddress.create( + self.api_client, + accountid=account.name, + zoneid=self.zone.id, + domainid=account.domainid, + networkid=isolated_network.id) + + FireWallRule.create( + self.apiclient, + ipaddressid=ipaddress.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=ipaddress.ipaddress.id, + networkid=isolated_network.id) try: - virtual_machine.get_ssh_client(ipaddress=ipaddress.ipaddress.ipaddress) + virtual_machine.get_ssh_client( + ipaddress=ipaddress.ipaddress.ipaddress) except Exception as e: - self.fail("Exception while SSHing to VM %s with IP %s" % (virtual_machine.id, ipaddress.ipaddress.ipaddress)) + self.fail( + "Exception while SSHing to VM %s with IP %s" % + (virtual_machine.id, ipaddress.ipaddress.ipaddress)) virtual_machine.delete(self.api_client) @@ -541,29 +730,38 @@ class TestPersistentNetworks(cloudstackTestCase): self.verifyVmExpungement(virtual_machine) # wait for time such that, network is cleaned up - wait_for_cleanup(self.api_client, ["network.gc.interval", "network.gc.wait"]) - - # Check network state now, this will bolster that network updation has taken effect - response = verifyNetworkState(self.api_client, isolated_network.id, "implemented") + wait_for_cleanup( + self.api_client, [ + "network.gc.interval", "network.gc.wait"]) + + # Check network state now, this will bolster that network updation has + # taken effect + response = verifyNetworkState( + self.api_client, + isolated_network.id, + "implemented") exceptionOccured = response[0] isNetworkInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccured or (not isNetworkInDesiredState)): - self.fail(exceptionMessage) + self.fail(exceptionMessage) return - @data("LB-VR","LB-Netscaler") + @data("LB-VR", "LB-Netscaler") @attr(tags=["advanced", "advancedns"]) def test_pf_nat_rule_persistent_network(self, value): - # This test shall run with two scenarios, one with LB services through VR and other through LB service + # This test shall run with two scenarios, one with LB services + # through VR and other through LB service # through Netscaler""" # steps - # 1. create isolated network with network offering which has ispersistent field enabled + # 1. create isolated network with network offering which has + # ispersistent field enabled # and LB service through VR or Netscaler - # 2. Check routers belonging to network and verify that router is reachable through host using linklocalip + # 2. Check routers belonging to network and verify that router + # is reachable through host using linklocalip # 3. Deploy VM ,acquire IP, create Firewall, NAT rules # 4. Verify the working of NAT, Firewall rules # @@ -575,27 +773,46 @@ class TestPersistentNetworks(cloudstackTestCase): networkOffering = self.isolated_persistent_network_offering if value == "LB-Netscaler": if self.ns_configured: - networkOffering = self.isolated_persistent_network_offering_netscaler + networkOffering = ( + self.isolated_persistent_network_offering_netscaler) else: - self.skipTest("Skipping - this test required netscaler configured in the network") + self.skipTest( + "Skipping - this test required netscaler\ + configured in the network") # Create account and network in it - account = Account.create(self.api_client, self.services["account"],domainid=self.domain.id) + account = Account.create( + self.api_client, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(account) - isolated_persistent_network = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=networkOffering.id, - accountid=account.name,domainid=self.domain.id, - zoneid=self.zone.id) - self.assertEqual(str(isolated_persistent_network.state).lower(), "implemented", "network state should be implemented, it is %s" \ - % isolated_persistent_network.state) - self.assertIsNotNone(isolated_persistent_network.vlan, "vlan must not be null for persistent network") + isolated_persistent_network = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=networkOffering.id, + accountid=account.name, + domainid=self.domain.id, + zoneid=self.zone.id) + self.assertEqual( + str( + isolated_persistent_network.state).lower(), + "implemented", + "network state should be implemented, it is %s" % + isolated_persistent_network.state) + self.assertIsNotNone( + isolated_persistent_network.vlan, + "vlan must not be null for persistent network") # Check if router is assigned to the persistent network routers = Router.list(self.api_client, account=account.name, domainid=account.domainid, networkid=isolated_persistent_network.id) - self.assertEqual(validateList(routers)[0], PASS, "Routers list validation failed, list is %s" % routers) + self.assertEqual( + validateList(routers)[0], + PASS, + "Routers list validation failed, list is %s" % + routers) router = routers[0] # Check if router if reachable from the host @@ -603,65 +820,109 @@ class TestPersistentNetworks(cloudstackTestCase): # Deploy VM in the network try: - virtual_machine = VirtualMachine.create(self.apiclient,self.services["virtual_machine"], - networkids=[isolated_persistent_network.id],serviceofferingid=self.service_offering.id, - accountid=account.name,domainid=self.domain.id) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + isolated_persistent_network.id], + serviceofferingid=self.service_offering.id, + accountid=account.name, + domainid=self.domain.id) except Exception as e: self.fail("vm creation failed: %s" % e) # Acquire IP address, create Firewall, NAT rule - self.debug("Associating public IP for network: %s" % isolated_persistent_network.id) - ipaddress = PublicIPAddress.create(self.api_client,accountid=account.name, - zoneid=self.zone.id,domainid=account.domainid, - networkid=isolated_persistent_network.id) - - FireWallRule.create(self.apiclient,ipaddressid=ipaddress.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=ipaddress.ipaddress.id, - networkid=isolated_persistent_network.id) + self.debug( + "Associating public IP for network: %s" % + isolated_persistent_network.id) + ipaddress = PublicIPAddress.create( + self.api_client, + accountid=account.name, + zoneid=self.zone.id, + domainid=account.domainid, + networkid=isolated_persistent_network.id) + + FireWallRule.create( + self.apiclient, + ipaddressid=ipaddress.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=ipaddress.ipaddress.id, + networkid=isolated_persistent_network.id) # Check working of PF, NAT rules try: - virtual_machine.get_ssh_client(ipaddress=ipaddress.ipaddress.ipaddress) + virtual_machine.get_ssh_client( + ipaddress=ipaddress.ipaddress.ipaddress) except Exception as e: - self.fail("Exception while SSHing to VM %s with IP %s" % (virtual_machine.id, ipaddress.ipaddress.ipaddress)) + self.fail( + "Exception while SSHing to VM %s with IP %s" % + (virtual_machine.id, ipaddress.ipaddress.ipaddress)) return @attr(tags=["advanced"]) def test_persistent_network_with_RVR(self): # steps - # 1. create account and isolated network with network offering which has ispersistent field enabled + # 1. create account and isolated network with network + # offering which has ispersistent field enabled # and supporting Redundant Virtual Router in it # 2. Check the Master and Backup Routers are present # 3. Deploy VM ,acquire IP, create Firewall, NAT rules # 4. Verify the working of NAT, Firewall rules # # validation - # 1. Two routers should belong to the network amd they should be reachable from host + # 1. Two routers should belong to the network and + # they should be reachable from host # 2. NAT and Firewall rule should work as expected # Create account and create isolated persistent network with RVR in it - account = Account.create(self.api_client, self.services["account"],domainid=self.domain.id) + account = Account.create( + self.api_client, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(account) - # Create isolated persistent network with RVR - isolated_persistent_network_RVR = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=self.isolated_persistent_network_offering_RVR.id, - accountid=account.name,domainid=self.domain.id, - zoneid=self.zone.id) - self.assertEqual(str(isolated_persistent_network_RVR.state).lower(), "implemented", "network state should be implemented, it is %s" \ - % isolated_persistent_network_RVR.state) - self.assertIsNotNone(isolated_persistent_network_RVR.vlan, "vlan must not be null for persistent network") + # Create isolated persistent network with RVR + isolated_persistent_network_RVR = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=self.isolated_persistent_network_offering_RVR.id, + accountid=account.name, + domainid=self.domain.id, + zoneid=self.zone.id) + self.assertEqual( + str( + isolated_persistent_network_RVR.state).lower(), + "implemented", + "network state should be implemented, it is %s" % + isolated_persistent_network_RVR.state) + self.assertIsNotNone( + isolated_persistent_network_RVR.vlan, + "vlan must not be null for persistent network") # Check if two routers belong to the network - self.debug("Listing routers for network: %s" % isolated_persistent_network_RVR.name) + self.debug( + "Listing routers for network: %s" % + isolated_persistent_network_RVR.name) routers = Router.list(self.api_client, listall=True, networkid=isolated_persistent_network_RVR.id) - self.assertEqual(validateList(routers)[0], PASS, "Routers list validation failed, list is %s" % routers) - self.assertEqual(len(routers), 2, "Length of the list router should be 2 (Backup & master)") + self.assertEqual( + validateList(routers)[0], + PASS, + "Routers list validation failed, list is %s" % + routers) + self.assertEqual( + len(routers), + 2, + "Length of the list router should be 2 (Backup & master)") # Check if routers are reachable from the host for router in routers: @@ -669,30 +930,51 @@ class TestPersistentNetworks(cloudstackTestCase): # Deploy VM try: - virtual_machine = VirtualMachine.create(self.apiclient,self.services["virtual_machine"], - networkids=[isolated_persistent_network_RVR.id],serviceofferingid=self.service_offering.id, - accountid=account.name,domainid=self.domain.id) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + isolated_persistent_network_RVR.id], + serviceofferingid=self.service_offering.id, + accountid=account.name, + domainid=self.domain.id) except Exception as e: self.fail("vm creation failed: %s" % e) # Acquire public ip, create Firewall, NAT rule - self.debug("Associating public IP for network: %s" % isolated_persistent_network_RVR.id) - ipaddress = PublicIPAddress.create(self.api_client,accountid=account.name, - zoneid=self.zone.id,domainid=account.domainid, - networkid=isolated_persistent_network_RVR.id) - - FireWallRule.create(self.apiclient,ipaddressid=ipaddress.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=ipaddress.ipaddress.id, - networkid=isolated_persistent_network_RVR.id) + self.debug( + "Associating public IP for network: %s" % + isolated_persistent_network_RVR.id) + ipaddress = PublicIPAddress.create( + self.api_client, + accountid=account.name, + zoneid=self.zone.id, + domainid=account.domainid, + networkid=isolated_persistent_network_RVR.id) + + FireWallRule.create( + self.apiclient, + ipaddressid=ipaddress.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=ipaddress.ipaddress.id, + networkid=isolated_persistent_network_RVR.id) # Check if Firewall, NAT rule work as expected try: - virtual_machine.get_ssh_client(ipaddress=ipaddress.ipaddress.ipaddress) + virtual_machine.get_ssh_client( + ipaddress=ipaddress.ipaddress.ipaddress) except Exception as e: - self.fail("Exception while SSHing to VM %s with IP %s" % (virtual_machine.id, ipaddress.ipaddress.ipaddress)) + self.fail( + "Exception while SSHing to VM %s with IP %s" % + (virtual_machine.id, ipaddress.ipaddress.ipaddress)) return @attr(tags=["advanced"]) @@ -705,93 +987,165 @@ class TestPersistentNetworks(cloudstackTestCase): # 1. VM should be deployed successfully in two networks # 2. All network rules should work as expected - account = Account.create(self.api_client, self.services["account"],domainid=self.domain.id) + account = Account.create( + self.api_client, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(account) - isolated_persistent_network_1 = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=self.isolated_persistent_network_offering.id, - accountid=account.name,domainid=self.domain.id, - zoneid=self.zone.id) - self.assertEqual(str(isolated_persistent_network_1.state).lower(), "implemented", "network state should be implemented, it is %s" \ - % isolated_persistent_network_1.state) - self.assertIsNotNone(isolated_persistent_network_1.vlan, "vlan must not be null for persistent network") - - isolated_persistent_network_2 = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=self.isolated_persistent_network_offering.id, - accountid=account.name,domainid=self.domain.id, - zoneid=self.zone.id) - self.assertEqual(str(isolated_persistent_network_2.state).lower(), "implemented", "network state should be implemented, it is %s" \ - % isolated_persistent_network_2.state) - self.assertIsNotNone(isolated_persistent_network_2.vlan, "vlan must not be null for persistent network") - - self.debug("Listing routers for network: %s" % isolated_persistent_network_1.name) + isolated_persistent_network_1 = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=self.isolated_persistent_network_offering.id, + accountid=account.name, + domainid=self.domain.id, + zoneid=self.zone.id) + self.assertEqual( + str( + isolated_persistent_network_1.state).lower(), + "implemented", + "network state should be implemented, it is %s" % + isolated_persistent_network_1.state) + self.assertIsNotNone( + isolated_persistent_network_1.vlan, + "vlan must not be null for persistent network") + + isolated_persistent_network_2 = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=self.isolated_persistent_network_offering.id, + accountid=account.name, + domainid=self.domain.id, + zoneid=self.zone.id) + self.assertEqual( + str( + isolated_persistent_network_2.state).lower(), + "implemented", + "network state should be implemented, it is %s" % + isolated_persistent_network_2.state) + self.assertIsNotNone( + isolated_persistent_network_2.vlan, + "vlan must not be null for persistent network") + + self.debug( + "Listing routers for network: %s" % + isolated_persistent_network_1.name) routers_nw_1 = Router.list(self.api_client, listall=True, - networkid=isolated_persistent_network_1.id) + networkid=isolated_persistent_network_1.id) - self.assertEqual(validateList(routers_nw_1)[0], PASS, "Routers list validation failed, list is %s" % routers_nw_1) + self.assertEqual( + validateList(routers_nw_1)[0], + PASS, + "Routers list validation failed, list is %s" % + routers_nw_1) # Check if router is reachable from the host for router in routers_nw_1: self.checkRouterAccessibility(router) - self.debug("Listing routers for network: %s" % isolated_persistent_network_2.name) + self.debug( + "Listing routers for network: %s" % + isolated_persistent_network_2.name) routers_nw_2 = Router.list(self.api_client, listall=True, - networkid=isolated_persistent_network_2.id) + networkid=isolated_persistent_network_2.id) - self.assertEqual(validateList(routers_nw_2)[0], PASS, "Routers list validation failed, list is %s" % routers_nw_2) + self.assertEqual( + validateList(routers_nw_2)[0], + PASS, + "Routers list validation failed, list is %s" % + routers_nw_2) # Check if router is reachable from the host for router in routers_nw_2: self.checkRouterAccessibility(router) try: - virtual_machine = VirtualMachine.create(self.apiclient,self.services["virtual_machine"], - networkids=[isolated_persistent_network_1.id, isolated_persistent_network_2.id], - serviceofferingid=self.service_offering.id, - accountid=account.name,domainid=self.domain.id) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + isolated_persistent_network_1.id, + isolated_persistent_network_2.id], + serviceofferingid=self.service_offering.id, + accountid=account.name, + domainid=self.domain.id) except Exception as e: self.fail("vm creation failed: %s" % e) - self.debug("Associating public IP for network: %s" % isolated_persistent_network_1.id) - ipaddress_nw_1 = PublicIPAddress.create(self.api_client,accountid=account.name, - zoneid=self.zone.id,domainid=account.domainid, - networkid=isolated_persistent_network_1.id) - - FireWallRule.create(self.apiclient,ipaddressid=ipaddress_nw_1.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - - self.debug("Associating public IP for network: %s" % isolated_persistent_network_2.id) - ipaddress_nw_2 = PublicIPAddress.create(self.api_client,accountid=account.name, - zoneid=self.zone.id,domainid=account.domainid, - networkid=isolated_persistent_network_2.id) - - FireWallRule.create(self.apiclient,ipaddressid=ipaddress_nw_2.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=ipaddress_nw_1.ipaddress.id, - networkid=isolated_persistent_network_1.id) + self.debug( + "Associating public IP for network: %s" % + isolated_persistent_network_1.id) + ipaddress_nw_1 = PublicIPAddress.create( + self.api_client, + accountid=account.name, + zoneid=self.zone.id, + domainid=account.domainid, + networkid=isolated_persistent_network_1.id) + + FireWallRule.create( + self.apiclient, + ipaddressid=ipaddress_nw_1.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + + self.debug( + "Associating public IP for network: %s" % + isolated_persistent_network_2.id) + ipaddress_nw_2 = PublicIPAddress.create( + self.api_client, + accountid=account.name, + zoneid=self.zone.id, + domainid=account.domainid, + networkid=isolated_persistent_network_2.id) + + FireWallRule.create( + self.apiclient, + ipaddressid=ipaddress_nw_2.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=ipaddress_nw_1.ipaddress.id, + networkid=isolated_persistent_network_1.id) try: - virtual_machine.get_ssh_client(ipaddress=ipaddress_nw_1.ipaddress.ipaddress) + virtual_machine.get_ssh_client( + ipaddress=ipaddress_nw_1.ipaddress.ipaddress) except Exception as e: - self.fail("Exception while SSHing to VM %s with IP %s" % (virtual_machine.id, ipaddress_nw_1.ipaddress.ipaddress)) + self.fail( + "Exception while SSHing to VM %s with IP %s" % + (virtual_machine.id, ipaddress_nw_1.ipaddress.ipaddress)) - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=ipaddress_nw_2.ipaddress.id, - networkid=isolated_persistent_network_2.id) + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=ipaddress_nw_2.ipaddress.id, + networkid=isolated_persistent_network_2.id) try: - virtual_machine.get_ssh_client(ipaddress=ipaddress_nw_2.ipaddress.ipaddress) + virtual_machine.get_ssh_client( + ipaddress=ipaddress_nw_2.ipaddress.ipaddress) except Exception as e: - self.fail("Exception while SSHing to VM %s with IP %s" % (virtual_machine.id, ipaddress_nw_2.ipaddress.ipaddress)) + self.fail( + "Exception while SSHing to VM %s with IP %s" % + (virtual_machine.id, ipaddress_nw_2.ipaddress.ipaddress)) return @attr(tags=["advanced"]) def test_vm_deployment_persistent_and_non_persistent_networks(self): # steps - # 1. create account and create two networks in it (persistent and non persistent) + # 1. create account and create two networks in it + # (persistent and non persistent) # 2. Deploy virtual machine in these two networks - # 3. Associate ip address with both the accounts and create firewall,port forwarding rule + # 3. Associate ip address with both the accounts and + # create firewall,port forwarding rule # 4. Try to SSH to VM through both the IPs # # validation @@ -799,62 +1153,110 @@ class TestPersistentNetworks(cloudstackTestCase): # 2. virtual machine should be created successfully # 3. SSH should be successful through both the IPs - account = Account.create(self.api_client, self.services["account"],domainid=self.domain.id) + account = Account.create( + self.api_client, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(account) - network_1 = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=self.isolated_persistent_network_offering.id, - accountid=account.name,domainid=self.domain.id, - zoneid=self.zone.id) - self.assertEqual(str(network_1.state).lower(), "implemented", "network state should be implemented, it is %s" \ - % network_1.state) - self.assertIsNotNone(network_1.vlan, "vlan must not be null for persistent network") - - network_2 = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=self.isolated_network_offering.id, - accountid=account.name,domainid=self.domain.id, - zoneid=self.zone.id) + network_1 = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=self.isolated_persistent_network_offering.id, + accountid=account.name, + domainid=self.domain.id, + zoneid=self.zone.id) + self.assertEqual( + str( + network_1.state).lower(), + "implemented", + "network state should be implemented, it is %s" % + network_1.state) + self.assertIsNotNone( + network_1.vlan, + "vlan must not be null for persistent network") + + network_2 = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=self.isolated_network_offering.id, + accountid=account.name, + domainid=self.domain.id, + zoneid=self.zone.id) try: - virtual_machine = VirtualMachine.create(self.apiclient,self.services["virtual_machine"], - networkids=[network_1.id, network_2.id], - serviceofferingid=self.service_offering.id, - accountid=account.name,domainid=self.domain.id) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network_1.id, + network_2.id], + serviceofferingid=self.service_offering.id, + accountid=account.name, + domainid=self.domain.id) except Exception as e: self.fail("vm creation failed: %s" % e) self.debug("Associating public IP for network: %s" % network_1.id) - ipaddress_nw_1 = PublicIPAddress.create(self.api_client,accountid=account.name, - zoneid=self.zone.id,domainid=account.domainid, - networkid=network_1.id) - - FireWallRule.create(self.apiclient,ipaddressid=ipaddress_nw_1.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=ipaddress_nw_1.ipaddress.id, - networkid=network_1.id) + ipaddress_nw_1 = PublicIPAddress.create( + self.api_client, + accountid=account.name, + zoneid=self.zone.id, + domainid=account.domainid, + networkid=network_1.id) + + FireWallRule.create( + self.apiclient, + ipaddressid=ipaddress_nw_1.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=ipaddress_nw_1.ipaddress.id, + networkid=network_1.id) try: - virtual_machine.get_ssh_client(ipaddress=ipaddress_nw_1.ipaddress.ipaddress) + virtual_machine.get_ssh_client( + ipaddress=ipaddress_nw_1.ipaddress.ipaddress) except Exception as e: - self.fail("Exception while SSHing to VM %s with IP %s" % (virtual_machine.id, ipaddress_nw_1.ipaddress.ipaddress)) + self.fail( + "Exception while SSHing to VM %s with IP %s" % + (virtual_machine.id, ipaddress_nw_1.ipaddress.ipaddress)) self.debug("Associating public IP for network: %s" % network_2.id) - ipaddress_nw_2 = PublicIPAddress.create(self.api_client,accountid=account.name, - zoneid=self.zone.id,domainid=account.domainid, - networkid=network_2.id) - - FireWallRule.create(self.apiclient,ipaddressid=ipaddress_nw_2.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=ipaddress_nw_2.ipaddress.id, - networkid=network_2.id) + ipaddress_nw_2 = PublicIPAddress.create( + self.api_client, + accountid=account.name, + zoneid=self.zone.id, + domainid=account.domainid, + networkid=network_2.id) + + FireWallRule.create( + self.apiclient, + ipaddressid=ipaddress_nw_2.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=ipaddress_nw_2.ipaddress.id, + networkid=network_2.id) try: - virtual_machine.get_ssh_client(ipaddress=ipaddress_nw_2.ipaddress.ipaddress) + virtual_machine.get_ssh_client( + ipaddress=ipaddress_nw_2.ipaddress.ipaddress) except Exception as e: - self.fail("Exception while SSHing to VM %s with IP %s" % (virtual_machine.id, ipaddress_nw_2.ipaddress.ipaddress)) + self.fail( + "Exception while SSHing to VM %s with IP %s" % + (virtual_machine.id, ipaddress_nw_2.ipaddress.ipaddress)) return @attr(tags=["advanced"]) @@ -873,45 +1275,75 @@ class TestPersistentNetworks(cloudstackTestCase): # 3. Network state should be allocated after network cleanup interval # Create account and create persistent network in it - account = Account.create(self.api_client, self.services["account"],domainid=self.domain.id) + account = Account.create( + self.api_client, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(account) - network = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=self.isolated_persistent_network_offering.id, - accountid=account.name,domainid=self.domain.id, - zoneid=self.zone.id) - self.assertEqual(str(network.state).lower(), "implemented", "network state should be implemented, it is %s" \ - % network.state) - self.assertIsNotNone(network.vlan, "vlan must not be null for persistent network") + network = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=self.isolated_persistent_network_offering.id, + accountid=account.name, + domainid=self.domain.id, + zoneid=self.zone.id) + self.assertEqual(str(network.state).lower(), + "implemented", + "network state should be implemented, it is %s" + % network.state) + self.assertIsNotNone( + network.vlan, + "vlan must not be null for persistent network") # Update network with non persistent network offering - network.update(self.apiclient, networkofferingid=self.isolated_network_offering.id) + network.update( + self.apiclient, + networkofferingid=self.isolated_network_offering.id) # Deploy VM, acquire IP, create NAT, firewall rules try: - virtual_machine = VirtualMachine.create(self.apiclient,self.services["virtual_machine"], - networkids=[network.id], - serviceofferingid=self.service_offering.id, - accountid=account.name,domainid=self.domain.id) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=account.name, + domainid=self.domain.id) except Exception as e: self.fail("vm creation failed: %s" % e) self.debug("Associating public IP for network: %s" % network.id) - ipaddress = PublicIPAddress.create(self.api_client,accountid=account.name, - zoneid=self.zone.id,domainid=account.domainid, - networkid=network.id) - - FireWallRule.create(self.apiclient,ipaddressid=ipaddress.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=ipaddress.ipaddress.id, - networkid=network.id) + ipaddress = PublicIPAddress.create( + self.api_client, + accountid=account.name, + zoneid=self.zone.id, + domainid=account.domainid, + networkid=network.id) + + FireWallRule.create( + self.apiclient, + ipaddressid=ipaddress.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=ipaddress.ipaddress.id, + networkid=network.id) # Verify working of network rules try: - virtual_machine.get_ssh_client(ipaddress=ipaddress.ipaddress.ipaddress) + virtual_machine.get_ssh_client( + ipaddress=ipaddress.ipaddress.ipaddress) except Exception as e: - self.fail("Exception while SSHing to VM %s with IP %s" % (virtual_machine.id, ipaddress.ipaddress.ipaddress)) + self.fail( + "Exception while SSHing to VM %s with IP %s" % + (virtual_machine.id, ipaddress.ipaddress.ipaddress)) # Delete VM virtual_machine.delete(self.api_client) @@ -920,16 +1352,19 @@ class TestPersistentNetworks(cloudstackTestCase): self.verifyVmExpungement(virtual_machine) # wait for time such that, network is cleaned up - wait_for_cleanup(self.api_client, ["network.gc.interval", "network.gc.wait"]) + wait_for_cleanup( + self.api_client, [ + "network.gc.interval", "network.gc.wait"]) - # Check network state now, this will bolster that network updation has taken effect + # Check network state now, this will bolster that network updation has + # taken effect response = verifyNetworkState(self.api_client, network.id, "allocated") exceptionOccured = response[0] isNetworkInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccured or (not isNetworkInDesiredState)): - self.fail(exceptionMessage) + self.fail(exceptionMessage) return @attr(tags=["advanced"]) @@ -940,59 +1375,100 @@ class TestPersistentNetworks(cloudstackTestCase): # 3. Delete the account in which network is created # # validation - # 1. Persistent network state should be implemented before VM creation and have some vlan assigned + # 1. Persistent network state should be implemented before VM creation + # and have some vlan assigned # 2. Network rules should work as expected # 3. Network and IPs should be freed after account is deleted - account = Account.create(self.api_client, self.services["account"],domainid=self.domain.id) - network = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=self.isolated_persistent_network_offering.id, - accountid=account.name,domainid=self.domain.id, - zoneid=self.zone.id) + account = Account.create( + self.api_client, + self.services["account"], + domainid=self.domain.id) + network = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=self.isolated_persistent_network_offering.id, + accountid=account.name, + domainid=self.domain.id, + zoneid=self.zone.id) try: - virtual_machine = VirtualMachine.create(self.apiclient,self.services["virtual_machine"], - networkids=[network.id],serviceofferingid=self.service_offering.id, - accountid=account.name,domainid=self.domain.id) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=account.name, + domainid=self.domain.id) except Exception as e: self.fail("vm creation failed: %s" % e) self.debug("Associating public IP for network: %s" % network.id) - ipaddress = PublicIPAddress.create(self.api_client,accountid=account.name, - zoneid=self.zone.id,domainid=account.domainid, - networkid=network.id) - - FireWallRule.create(self.apiclient,ipaddressid=ipaddress.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=ipaddress.ipaddress.id, - networkid=network.id) + ipaddress = PublicIPAddress.create( + self.api_client, + accountid=account.name, + zoneid=self.zone.id, + domainid=account.domainid, + networkid=network.id) + + FireWallRule.create( + self.apiclient, + ipaddressid=ipaddress.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=ipaddress.ipaddress.id, + networkid=network.id) try: - virtual_machine.get_ssh_client(ipaddress=ipaddress.ipaddress.ipaddress) + virtual_machine.get_ssh_client( + ipaddress=ipaddress.ipaddress.ipaddress) except Exception as e: - self.fail("Exception while SSHing to VM %s with IP %s" % (virtual_machine.id, ipaddress.ipaddress.ipaddress)) + self.fail( + "Exception while SSHing to VM %s with IP %s" % + (virtual_machine.id, ipaddress.ipaddress.ipaddress)) # Delete the account account.delete(self.api_client) # Verify the resources belonging to account have been cleaned up networks = Network.list(self.apiclient, id=network.id) - self.assertEqual(validateList(networks)[0], FAIL, "network list should be enmpty, it is %s" % networks) - - public_ips = PublicIPAddress.list(self.apiclient, id=ipaddress.ipaddress.id) - self.assertEqual(validateList(public_ips)[0], FAIL, "Public Ip list be empty, it is %s" % public_ips) + self.assertEqual( + validateList(networks)[0], + FAIL, + "network list should be enmpty, it is %s" % + networks) + + public_ips = PublicIPAddress.list( + self.apiclient, + id=ipaddress.ipaddress.id) + self.assertEqual( + validateList(public_ips)[0], + FAIL, + "Public Ip list be empty, it is %s" % + public_ips) return + @ddt class TestAssignVirtualMachine(cloudstackTestCase): - """Test Persistent Network creation with assigning VM to different account/domain + + """Test Persistent Network creation with + assigning VM to different account/domain """ @classmethod def setUpClass(cls): - cls.testClient = super(TestAssignVirtualMachine, cls).getClsTestClient() + cls.testClient = super( + TestAssignVirtualMachine, + cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() # Fill services from the external config file @@ -1002,39 +1478,48 @@ class TestAssignVirtualMachine(cloudstackTestCase): cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.template = get_template( - cls.api_client, - cls.zone.id, - cls.services["ostype"] - ) + cls.api_client, + cls.zone.id, + cls.services["ostype"] + ) if cls.template == FAILED: - assert False, "get_template() failed to return template with description %s" % cls.services["ostype"] + assert False, "get_template() failed to return template\ + with description %s" % 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.isolated_persistent_network_offering = cls.createNetworkOffering("nw_off_isolated_persistent") - cls.isolated_persistent_network_offering_RVR = cls.createNetworkOffering("nw_off_persistent_RVR") - cls.persistent_network_offering_netscaler = cls.createNetworkOffering("nw_off_isolated_persistent_netscaler") + cls.api_client, + cls.services["service_offering"] + ) + cls.isolated_persistent_network_offering = cls.createNetworkOffering( + "nw_off_isolated_persistent") + cls.isolated_persistent_network_offering_RVR =\ + cls.createNetworkOffering( + "nw_off_persistent_RVR" + ) + cls.persistent_network_offering_netscaler = cls.createNetworkOffering( + "nw_off_isolated_persistent_netscaler") # Configure Netscaler device - # If configuration succeeds, set ns_configured to True so that Netscaler tests are executed + # If configuration succeeds, set ns_configured to True so that + # Netscaler tests are executed cls.ns_configured = False try: - cls.netscaler = add_netscaler(cls.api_client, cls.zone.id, cls.services["netscaler_VPX"]) - cls._cleanup.append(cls.netscaler) - cls.ns_configured = True + cls.netscaler = add_netscaler( + cls.api_client, + cls.zone.id, + cls.services["netscaler_VPX"]) + cls._cleanup.append(cls.netscaler) + cls.ns_configured = True except Exception: - cls.ns_configured = False - + cls.ns_configured = False # network will be deleted as part of account cleanup cls._cleanup = [ - cls.service_offering, cls.isolated_persistent_network_offering, - cls.isolated_persistent_network_offering_RVR, - cls.persistent_network_offering_netscaler - ] + cls.service_offering, cls.isolated_persistent_network_offering, + cls.isolated_persistent_network_offering_RVR, + cls.persistent_network_offering_netscaler + ] return @classmethod @@ -1049,19 +1534,22 @@ class TestAssignVirtualMachine(cloudstackTestCase): @classmethod def createNetworkOffering(cls, network_offering_type): network_offering = NetworkOffering.create( - cls.api_client, - cls.services[network_offering_type], - conservemode=False - ) + cls.api_client, + cls.services[network_offering_type], + conservemode=False + ) # Update network offering state from disabled to enabled. - NetworkOffering.update(network_offering, cls.api_client, id=network_offering.id, - state="enabled") + NetworkOffering.update( + network_offering, +
<TRUNCATED>
