Return-Path: X-Original-To: apmail-cloudstack-commits-archive@www.apache.org Delivered-To: apmail-cloudstack-commits-archive@www.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id DEB3C172CC for ; Mon, 23 Mar 2015 10:27:41 +0000 (UTC) Received: (qmail 92479 invoked by uid 500); 23 Mar 2015 10:27:41 -0000 Delivered-To: apmail-cloudstack-commits-archive@cloudstack.apache.org Received: (qmail 92438 invoked by uid 500); 23 Mar 2015 10:27:41 -0000 Mailing-List: contact commits-help@cloudstack.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@cloudstack.apache.org Delivered-To: mailing list commits@cloudstack.apache.org Received: (qmail 92426 invoked by uid 99); 23 Mar 2015 10:27:41 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 23 Mar 2015 10:27:41 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id 7C048E0FB5; Mon, 23 Mar 2015 10:27:41 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: talluri@apache.org To: commits@cloudstack.apache.org Date: Mon, 23 Mar 2015 10:27:41 -0000 Message-Id: <0ed2dc70f36f4909927efe5b58cb8d74@git.apache.org> X-Mailer: ASF-Git Admin Mailer Subject: [1/2] git commit: updated refs/heads/master to c094274 Repository: cloudstack Updated Branches: refs/heads/master 6a881b933 -> c09427401 http://git-wip-us.apache.org/repos/asf/cloudstack/blob/c0942740/test/integration/component/test_vpc_network.py ---------------------------------------------------------------------- diff --git a/test/integration/component/test_vpc_network.py b/test/integration/component/test_vpc_network.py index e183054..cbc5eea 100644 --- a/test/integration/component/test_vpc_network.py +++ b/test/integration/component/test_vpc_network.py @@ -17,36 +17,40 @@ """ Component tests for VPC network functionality - with and without Netscaler (Netscaler tests will be skipped if Netscaler configuration fails) """ -#Import Local Modules +# Import Local Modules from nose.plugins.attrib import attr from marvin.cloudstackTestCase import cloudstackTestCase, unittest from marvin.cloudstackAPI import startVirtualMachine, stopVirtualMachine from marvin.lib.utils import cleanup_resources, validateList from marvin.lib.base import (VirtualMachine, - ServiceOffering, - Account, - NATRule, - NetworkOffering, - Network, - VPC, - VpcOffering, - LoadBalancerRule, - Router, - StaticNATRule, - NetworkACL, - PublicIPAddress) + ServiceOffering, + Account, + NATRule, + NetworkOffering, + Network, + VPC, + VpcOffering, + LoadBalancerRule, + Router, + StaticNATRule, + NetworkACL, + PublicIPAddress) from marvin.lib.common import (get_zone, - get_domain, - get_template, - wait_for_cleanup, - add_netscaler, - list_networks) -# For more info on ddt refer to http://ddt.readthedocs.org/en/latest/api.html#module-ddt + get_domain, + get_template, + wait_for_cleanup, + add_netscaler, + list_networks, + verifyRouterState) +# For more info on ddt refer to +# http://ddt.readthedocs.org/en/latest/api.html#module-ddt from ddt import ddt, data import time from marvin.codes import PASS + class Services: + """Test VPC network services """ @@ -91,7 +95,8 @@ class Services: "SourceNat": {"SupportedSourceNatTypes": "peraccount"}, }, }, - # Offering that uses Netscaler as provider for LB inside VPC, dedicated = false + # Offering that uses Netscaler as provider for LB inside VPC, + # dedicated = false "network_off_netscaler": { "name": 'Network offering-netscaler', "displaytext": 'Network offering-netscaler', @@ -117,33 +122,33 @@ class Services: # Offering that uses Netscaler as provider for LB in VPC, dedicated = True # This offering is required for the tests that use Netscaler as external LB provider in VPC "network_offering_vpcns": { - "name": 'VPC Network offering', - "displaytext": 'VPC Network off', - "guestiptype": 'Isolated', - "supportedservices": 'Vpn,Dhcp,Dns,SourceNat,PortForwarding,Lb,UserData,StaticNat,NetworkACL', - "traffictype": 'GUEST', - "availability": 'Optional', - "useVpc": 'on', - "serviceProviderList": { - "Vpn": 'VpcVirtualRouter', - "Dhcp": 'VpcVirtualRouter', - "Dns": 'VpcVirtualRouter', - "SourceNat": 'VpcVirtualRouter', - "PortForwarding": 'VpcVirtualRouter', - "Lb": 'Netscaler', - "UserData": 'VpcVirtualRouter', - "StaticNat": 'VpcVirtualRouter', - "NetworkACL": 'VpcVirtualRouter' - }, - "serviceCapabilityList": { - "SourceNat": { - "SupportedSourceNatTypes": "peraccount" - }, - "lb": { - "SupportedLbIsolation": "dedicated" - }, - }, - }, + "name": 'VPC Network offering', + "displaytext": 'VPC Network off', + "guestiptype": 'Isolated', + "supportedservices": 'Vpn,Dhcp,Dns,SourceNat,PortForwarding,Lb,UserData,StaticNat,NetworkACL', + "traffictype": 'GUEST', + "availability": 'Optional', + "useVpc": 'on', + "serviceProviderList": { + "Vpn": 'VpcVirtualRouter', + "Dhcp": 'VpcVirtualRouter', + "Dns": 'VpcVirtualRouter', + "SourceNat": 'VpcVirtualRouter', + "PortForwarding": 'VpcVirtualRouter', + "Lb": 'Netscaler', + "UserData": 'VpcVirtualRouter', + "StaticNat": 'VpcVirtualRouter', + "NetworkACL": 'VpcVirtualRouter' + }, + "serviceCapabilityList": { + "SourceNat": { + "SupportedSourceNatTypes": "peraccount" + }, + "lb": { + "SupportedLbIsolation": "dedicated" + }, + }, + }, "network_off_shared": { "name": 'Shared Network offering', @@ -167,16 +172,16 @@ class Services: }, # Netscaler should be added as a dedicated device for it to work as external LB provider in VPC "netscaler": { - "ipaddress": '10.102.192.50', - "username": 'nsroot', - "password": 'nsroot', - "networkdevicetype": 'NetscalerVPXLoadBalancer', - "publicinterface": '1/3', - "privateinterface": '1/4', - "numretries": 2, - "lbdevicededicated": True, - "lbdevicecapacity": 50, - "port": 22, + "ipaddress": '10.102.192.50', + "username": 'nsroot', + "password": 'nsroot', + "networkdevicetype": 'NetscalerVPXLoadBalancer', + "publicinterface": '1/3', + "privateinterface": '1/4', + "numretries": 2, + "lbdevicededicated": True, + "lbdevicecapacity": 50, + "port": 22, }, "network": { "name": "Test Network", @@ -234,6 +239,7 @@ class Services: "timeout": 10, } + @ddt class TestVPCNetwork(cloudstackTestCase): @@ -245,40 +251,45 @@ class TestVPCNetwork(cloudstackTestCase): cls.services = Services().services # Added an attribute to track if Netscaler addition was successful. - # Value is checked in tests and if not configured, Netscaler tests will be skipped + # Value is checked in tests and if not configured, Netscaler tests will + # be skipped cls.ns_configured = False # Get Zone, Domain and templates 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.services["virtual_machine"]["zoneid"] = cls.zone.id cls.services["virtual_machine"]["template"] = cls.template.id cls._cleanup = [] cls.service_offering = ServiceOffering.create( - cls.api_client, - cls.services["service_offering"] - ) + cls.api_client, + cls.services["service_offering"] + ) cls._cleanup.append(cls.service_offering) # 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 try: - cls.netscaler = add_netscaler(cls.api_client, cls.zone.id, cls.services["netscaler"]) - cls._cleanup.append(cls.netscaler) - cls.debug("Netscaler configured") - cls.ns_configured = True + cls.netscaler = add_netscaler( + cls.api_client, + cls.zone.id, + cls.services["netscaler"]) + cls._cleanup.append(cls.netscaler) + cls.debug("Netscaler configured") + cls.ns_configured = True except Exception as e: - cls.debug("Warning: Couldn't configure Netscaler: %s" % e) + cls.debug("Warning: Couldn't configure Netscaler: %s" % e) 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) @@ -289,11 +300,11 @@ class TestVPCNetwork(cloudstackTestCase): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.account = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) + self.apiclient, + self.services["account"], + admin=True, + domainid=self.domain.id + ) self.cleanup = [self.account, ] return @@ -302,7 +313,6 @@ class TestVPCNetwork(cloudstackTestCase): cleanup_resources(self.apiclient, self.cleanup) except Exception as e: self.debug("Warning: Exception during cleanup : %s" % e) - #raise Exception("Warning: Exception during cleanup : %s" % e) return def validate_vpc_offering(self, vpc_offering): @@ -310,22 +320,22 @@ class TestVPCNetwork(cloudstackTestCase): self.debug("Check if the VPC offering is created successfully?") vpc_offs = VpcOffering.list( - self.apiclient, - id=vpc_offering.id - ) + self.apiclient, + id=vpc_offering.id + ) self.assertEqual( - isinstance(vpc_offs, list), - True, - "List VPC offerings should return a valid list" - ) + isinstance(vpc_offs, list), + True, + "List VPC offerings should return a valid list" + ) self.assertEqual( - vpc_offering.name, - vpc_offs[0].name, - "Name of the VPC offering should match with listVPCOff data" - ) + vpc_offering.name, + vpc_offs[0].name, + "Name of the VPC offering should match with listVPCOff data" + ) self.debug( - "VPC offering is created successfully - %s" % - vpc_offering.name) + "VPC offering is created successfully - %s" % + vpc_offering.name) return def validate_vpc_network(self, network, state=None): @@ -333,25 +343,25 @@ class TestVPCNetwork(cloudstackTestCase): self.debug("Check if the VPC network is created successfully?") vpc_networks = VPC.list( - self.apiclient, - id=network.id - ) + self.apiclient, + id=network.id + ) self.assertEqual( - isinstance(vpc_networks, list), - True, - "List VPC network should return a valid list" - ) + isinstance(vpc_networks, list), + True, + "List VPC network should return a valid list" + ) self.assertEqual( - network.name, - vpc_networks[0].name, - "Name of the VPC network should match with listVPC data" - ) + network.name, + vpc_networks[0].name, + "Name of the VPC network should match with listVPC data" + ) if state: self.assertEqual( - vpc_networks[0].state, - state, + vpc_networks[0].state, + state, "VPC state should be '%s'" % state - ) + ) self.debug("VPC network validated - %s" % network.name) return @@ -367,88 +377,89 @@ class TestVPCNetwork(cloudstackTestCase): # all of supported Services(Vpn,dhcpdns,UserData, SourceNat,Static # NAT,LB and PF,LB,NetworkAcl ) provided by VPCVR and conserve # mode is ON - # 3. Create a network tier using the network offering created in step2 as - # part of this VPC. + # 3. Create a network tier using the network offering created in + # step 2 as part of this VPC. # 4. Validate Network is created - # 5. Repeat test for offering which has Netscaler as external LB provider + # 5. Repeat test for offering which has Netscaler as external LB + # provider - if (value == "network_offering_vpcns" and self.ns_configured == False): - self.skipTest('Netscaler not configured: skipping test') + if (value == "network_offering_vpcns" and not self.ns_configured): + self.skipTest('Netscaler not configured: skipping test') if (value == "network_offering"): - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering', - listall=True - ) + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering', + listall=True + ) else: - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering with Netscaler', - listall=True - ) - vpc_off=vpc_off_list[0] + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering with Netscaler', + listall=True + ) + vpc_off = vpc_off_list[0] self.debug("Creating a VPC with offering: %s" % vpc_off.id) self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.network_offering = NetworkOffering.create( - self.apiclient, - self.services[value], - conservemode=False - ) + self.apiclient, + self.services[value], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) # Creating network using the network offering created self.debug("Creating network with network offering: %s" % - self.network_offering.id) + self.network_offering.id) network = Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.1.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.1.1', + vpcid=vpc.id + ) self.debug("Created network with ID: %s" % network.id) self.debug( "Verifying list network response to check if network created?") networks = Network.list( - self.apiclient, - id=network.id, - listall=True - ) + self.apiclient, + id=network.id, + listall=True + ) self.assertEqual( - isinstance(networks, list), - True, - "List networks should return a valid response" - ) + isinstance(networks, list), + True, + "List networks should return a valid response" + ) nw = networks[0] self.assertEqual( nw.networkofferingid, self.network_offering.id, "Network should be created from network offering - %s" % - self.network_offering.id - ) + self.network_offering.id + ) self.assertEqual( - nw.vpcid, - vpc.id, - "Network should be created in VPC: %s" % vpc.name - ) + nw.vpcid, + vpc.id, + "Network should be created in VPC: %s" % vpc.name + ) return @data("network_offering", "network_offering_vpcns") @@ -465,65 +476,66 @@ class TestVPCNetwork(cloudstackTestCase): # and conserve mode is ON # 3. Create a network using the network offering created in step2 as # part of this VPC. - # 4. Network creation should fail since SourceNat offered by VR instead of VPCVR - # 5. Repeat test for offering which has Netscaler as external LB provider - - if (value == "network_offering_vpcns" and self.ns_configured == False): - self.skipTest('Netscaler not configured: skipping test') + # 4. Network creation should fail since SourceNat offered by VR + # instead of VPCVR + # 5. Repeat test for offering which has Netscaler as external LB + # provider + if (value == "network_offering_vpcns" and not self.ns_configured): + self.skipTest('Netscaler not configured: skipping test') if (value == "network_offering"): - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering', - listall=True - ) + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering', + listall=True + ) else: - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering with Netscaler', - listall=True - ) - vpc_off=vpc_off_list[0] + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering with Netscaler', + listall=True + ) + vpc_off = vpc_off_list[0] self.debug("Creating a VPC with offering: %s" % vpc_off.id) self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.services[value]["serviceProviderList"] = { - "SourceNat": 'VirtualRouter', } + "SourceNat": 'VirtualRouter', } self.network_offering = NetworkOffering.create( - self.apiclient, - self.services[value], - conservemode=False - ) + self.apiclient, + self.services[value], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) # Creating network using the network offering created self.debug("Creating network with network offering: %s" % - self.network_offering.id) + self.network_offering.id) with self.assertRaises(Exception): Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.1.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.1.1', + vpcid=vpc.id + ) return @data("network_offering", "network_offering_vpcns") @@ -541,99 +553,100 @@ class TestVPCNetwork(cloudstackTestCase): # 4. Create another network using the network offering created in # step3 as part of this VPC # 5. Create Network should fail - # 6. Repeat test for offering which has Netscaler as external LB provider - if (value == "network_offering_vpcns" and self.ns_configured == False): - self.skipTest('Netscaler not configured: skipping test') - + # 6. Repeat test for offering which has Netscaler as external LB + # provider + if (value == "network_offering_vpcns" and not self.ns_configured): + self.skipTest('Netscaler not configured: skipping test') if (value == "network_offering"): - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering', - listall=True - ) + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering', + listall=True + ) else: - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering with Netscaler', - listall=True - ) - vpc_off=vpc_off_list[0] + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering with Netscaler', + listall=True + ) + vpc_off = vpc_off_list[0] self.debug("Creating a VPC with offering: %s" % vpc_off.id) self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.network_offering = NetworkOffering.create( - self.apiclient, - self.services[value], - conservemode=False - ) + self.apiclient, + self.services[value], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) # Creating network using the network offering created self.debug("Creating network with network offering: %s" % - self.network_offering.id) + self.network_offering.id) network = Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.1.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.1.1', + vpcid=vpc.id + ) self.debug("Created network with ID: %s" % network.id) self.debug( "Verifying list network response to check if network created?") networks = Network.list( - self.apiclient, - id=network.id, - listall=True - ) + self.apiclient, + id=network.id, + listall=True + ) self.assertEqual( - isinstance(networks, list), - True, - "List networks should return a valid response" - ) + isinstance(networks, list), + True, + "List networks should return a valid response" + ) nw = networks[0] self.assertEqual( nw.networkofferingid, self.network_offering.id, "Network should be created from network offering - %s" % - self.network_offering.id - ) + self.network_offering.id + ) self.assertEqual( - nw.vpcid, - vpc.id, - "Network should be created in VPC: %s" % vpc.name - ) + nw.vpcid, + vpc.id, + "Network should be created in VPC: %s" % vpc.name + ) self.debug("Creating another network in VPC: %s" % vpc.name) with self.assertRaises(Exception): Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.2.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.2.1', + vpcid=vpc.id + ) self.debug( - "Network creation failed as network with LB service already exists") + "Network creation failed as network with LB service\ + already exists") return @attr(tags=["intervlan"]) @@ -648,48 +661,48 @@ class TestVPCNetwork(cloudstackTestCase): # 3. Create a network using this network offering as part of this VPC. # 4. Create Network should fail since it doesn't match the VPC offering - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering', - listall=True - ) - vpc_off=vpc_off_list[0] + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering', + listall=True + ) + vpc_off = vpc_off_list[0] self.debug("Creating a VPC with offering: %s" % vpc_off.id) self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.network_offering = NetworkOffering.create( - self.apiclient, - self.services["network_offering_vpcns"], - conservemode=False - ) + self.apiclient, + self.services["network_offering_vpcns"], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) # Creating network using the network offering created self.debug("Creating network with network offering: %s" % - self.network_offering.id) + self.network_offering.id) with self.assertRaises(Exception): - Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.1.1', - vpcid=vpc.id - ) + Network.create( + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.1.1', + vpcid=vpc.id + ) self.debug("Network creation failed") return @@ -712,9 +725,9 @@ class TestVPCNetwork(cloudstackTestCase): self.debug("Creating a VPC offering..") vpc_off = VpcOffering.create( - self.apiclient, - self.services["vpc_offering"] - ) + self.apiclient, + self.services["vpc_offering"] + ) self.cleanup.append(vpc_off) self.validate_vpc_offering(vpc_off) @@ -723,48 +736,48 @@ class TestVPCNetwork(cloudstackTestCase): vpc_off.update(self.apiclient, state='Enabled') self.debug("creating a VPC network in the account: %s" % - self.account.name) + self.account.name) self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) # Enable redundant router capability for the network offering self.services["network"]["serviceCapabilityList"] = { - "SourceNat": { - "RedundantRouter": "true", - }, - } + "SourceNat": { + "RedundantRouter": "true", + }, + } self.network_offering = NetworkOffering.create( - self.apiclient, - self.services["network_offering"], - conservemode=False - ) + self.apiclient, + self.services["network_offering"], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) # Creating network using the network offering created self.debug("Creating network with network offering: %s" % - self.network_offering.id) + self.network_offering.id) with self.assertRaises(Exception): Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.2.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.2.1', + vpcid=vpc.id + ) self.debug("Network creation failed") return @@ -783,12 +796,13 @@ class TestVPCNetwork(cloudstackTestCase): # 5. Create network fails since VPC offering doesn't support LB self.debug("Creating a VPC offering without LB service") - self.services["vpc_offering"]["supportedservices"] = 'Dhcp,Dns,SourceNat,PortForwarding,Vpn,UserData,StaticNat' + self.services["vpc_offering"][ + "supportedservices"] = 'Dhcp,Dns,SourceNat,PortForwarding,Vpn,UserData,StaticNat' vpc_off = VpcOffering.create( - self.apiclient, - self.services["vpc_offering"] - ) + self.apiclient, + self.services["vpc_offering"] + ) self.cleanup.append(vpc_off) self.validate_vpc_offering(vpc_off) @@ -797,41 +811,41 @@ class TestVPCNetwork(cloudstackTestCase): vpc_off.update(self.apiclient, state='Enabled') self.debug("creating a VPC network in the account: %s" % - self.account.name) + self.account.name) self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.network_offering = NetworkOffering.create( - self.apiclient, - self.services["network_offering"], - conservemode=False - ) + self.apiclient, + self.services["network_offering"], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) # Creating network using the network offering created self.debug("Creating network with network offering: %s" % - self.network_offering.id) + self.network_offering.id) with self.assertRaises(Exception): Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.2.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.2.1', + vpcid=vpc.id + ) self.debug("Network creation failed as VPC doesn't have LB service") return @@ -852,12 +866,13 @@ class TestVPCNetwork(cloudstackTestCase): # part of this VPC self.debug("Creating a VPC offering without LB service") - self.services["vpc_offering"]["supportedservices"] = 'Dhcp,Dns,SourceNat,PortForwarding,UserData,StaticNat' + self.services["vpc_offering"][ + "supportedservices"] = 'Dhcp,Dns,SourceNat,PortForwarding,UserData,StaticNat' vpc_off = VpcOffering.create( - self.apiclient, - self.services["vpc_offering"] - ) + self.apiclient, + self.services["vpc_offering"] + ) self.cleanup.append(vpc_off) self.validate_vpc_offering(vpc_off) @@ -866,37 +881,38 @@ class TestVPCNetwork(cloudstackTestCase): vpc_off.update(self.apiclient, state='Enabled') self.debug("creating a VPC network in the account: %s" % - self.account.name) + self.account.name) self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.debug("Creating network offering without SourceNAT service") - self.services["network_offering"]["supportedservices"] = 'Dhcp,Dns,PortForwarding,Lb,UserData,StaticNat,NetworkACL' + self.services["network_offering"][ + "supportedservices"] = 'Dhcp,Dns,PortForwarding,Lb,UserData,StaticNat,NetworkACL' self.services["network_offering"]["serviceProviderList"] = { - "Dhcp": 'VpcVirtualRouter', - "Dns": 'VpcVirtualRouter', - "PortForwarding": 'VpcVirtualRouter', - "Lb": 'VpcVirtualRouter', - "UserData": 'VpcVirtualRouter', - "StaticNat": 'VpcVirtualRouter', - "NetworkACL": 'VpcVirtualRouter' - } + "Dhcp": 'VpcVirtualRouter', + "Dns": 'VpcVirtualRouter', + "PortForwarding": 'VpcVirtualRouter', + "Lb": 'VpcVirtualRouter', + "UserData": 'VpcVirtualRouter', + "StaticNat": 'VpcVirtualRouter', + "NetworkACL": 'VpcVirtualRouter' + } self.debug("Creating network offering without SourceNAT") with self.assertRaises(Exception): NetworkOffering.create( - self.apiclient, - self.services["network_offering"], - conservemode=False - ) + self.apiclient, + self.services["network_offering"], + conservemode=False + ) self.debug("Network creation failed as VPC doesn't have LB service") return @@ -910,46 +926,48 @@ class TestVPCNetwork(cloudstackTestCase): # 1. Create VPC Offering using Default Offering # 2. Create a VPC using the above VPC offering # 3. Create a network offering with guest type=shared - # 4. Create a network using the network offering created in step3 as part of this VPC + # 4. Create a network using the network offering created + # in step3 as part of this VPC # 5. Create network fails since it using shared offering - # 6. Repeat test for offering which has Netscaler as external LB provider + # 6. Repeat test for offering which has Netscaler as external LB + # provider - if (value == "network_offering_vpcns" and self.ns_configured == False): - self.skipTest('Netscaler not configured: skipping test') + if (value == "network_offering_vpcns" and not self.ns_configured): + self.skipTest('Netscaler not configured: skipping test') if (value == "network_off_shared"): - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering', - listall=True - ) + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering', + listall=True + ) else: - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering with Netscaler', - listall=True - ) - vpc_off=vpc_off_list[0] + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering with Netscaler', + listall=True + ) + vpc_off = vpc_off_list[0] self.debug("Creating a VPC with offering: %s" % vpc_off.id) self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.debug("Creating network offering with guesttype=shared") self.network_offering = NetworkOffering.create( - self.apiclient, - self.services["network_off_shared"], - conservemode=False - ) + self.apiclient, + self.services["network_off_shared"], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) @@ -957,18 +975,18 @@ class TestVPCNetwork(cloudstackTestCase): # Creating network using the network offering created self.debug( "Creating network with network offering without SourceNAT: %s" % - self.network_offering.id) + self.network_offering.id) with self.assertRaises(Exception): Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.1.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.1.1', + vpcid=vpc.id + ) self.debug("Network creation failed") return @@ -982,21 +1000,24 @@ class TestVPCNetwork(cloudstackTestCase): # 1. Create a network offering with guest type=Isolated that has all # supported Services(Vpn,dhcpdns,UserData, SourceNat,Static NAT,LB # and PF,LB,NetworkAcl ) provided by VPCVR and conserve mode is ON - # 2. Create offering fails since Conserve mode ON isn't allowed within VPC - # 3. Repeat test for offering which has Netscaler as external LB provider + # 2. Create offering fails since Conserve mode ON isn't allowed within + # VPC + # 3. Repeat test for offering which has Netscaler as external LB + # provider self.debug("Creating network offering with conserve mode = ON") with self.assertRaises(Exception): NetworkOffering.create( - self.apiclient, - self.services[value], - conservemode=True - ) + self.apiclient, + self.services[value], + conservemode=True + ) self.debug( - "Network creation failed as VPC support nw with conserve mode OFF") + "Network creation failed as VPC support nw with conserve mode OFF") return + @ddt class TestVPCNetworkRanges(cloudstackTestCase): @@ -1008,39 +1029,44 @@ class TestVPCNetworkRanges(cloudstackTestCase): cls.services = Services().services # Added an attribute to track if Netscaler addition was successful. - # Value is checked in tests and if not configured, Netscaler tests will be skipped + # Value is checked in tests and if not configured, Netscaler tests will + # be skipped cls.ns_configured = False # Get Zone, Domain and templates 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.services["virtual_machine"]["zoneid"] = cls.zone.id cls.services["virtual_machine"]["template"] = cls.template.id cls._cleanup = [] cls.service_offering = ServiceOffering.create( - cls.api_client, - cls.services["service_offering"] - ) + cls.api_client, + cls.services["service_offering"] + ) cls._cleanup.append(cls.service_offering) # 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 try: - cls.netscaler = add_netscaler(cls.api_client, cls.zone.id, cls.services["netscaler"]) - cls._cleanup.append(cls.netscaler) - cls.ns_configured = True + cls.netscaler = add_netscaler( + cls.api_client, + cls.zone.id, + cls.services["netscaler"]) + cls._cleanup.append(cls.netscaler) + cls.ns_configured = True except Exception as e: - cls.debug("Warning: Couldn't configure Netscaler: %s" % e) + cls.debug("Warning: Couldn't configure Netscaler: %s" % e) 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) @@ -1050,11 +1076,11 @@ class TestVPCNetworkRanges(cloudstackTestCase): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.account = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) + self.apiclient, + self.services["account"], + admin=True, + domainid=self.domain.id + ) self.cleanup = [self.account, ] return @@ -1070,22 +1096,22 @@ class TestVPCNetworkRanges(cloudstackTestCase): self.debug("Check if the VPC offering is created successfully?") vpc_offs = VpcOffering.list( - self.apiclient, - id=vpc_offering.id - ) + self.apiclient, + id=vpc_offering.id + ) self.assertEqual( - isinstance(vpc_offs, list), - True, - "List VPC offerings should return a valid list" - ) + isinstance(vpc_offs, list), + True, + "List VPC offerings should return a valid list" + ) self.assertEqual( - vpc_offering.name, - vpc_offs[0].name, - "Name of the VPC offering should match with listVPCOff data" - ) + vpc_offering.name, + vpc_offs[0].name, + "Name of the VPC offering should match with listVPCOff data" + ) self.debug( - "VPC offering is created successfully - %s" % - vpc_offering.name) + "VPC offering is created successfully - %s" % + vpc_offering.name) return def validate_vpc_network(self, network, state=None): @@ -1093,25 +1119,25 @@ class TestVPCNetworkRanges(cloudstackTestCase): self.debug("Check if the VPC network is created successfully?") vpc_networks = VPC.list( - self.apiclient, - id=network.id - ) + self.apiclient, + id=network.id + ) self.assertEqual( - isinstance(vpc_networks, list), - True, - "List VPC network should return a valid list" - ) + isinstance(vpc_networks, list), + True, + "List VPC network should return a valid list" + ) self.assertEqual( - network.name, - vpc_networks[0].name, - "Name of the VPC network should match with listVPC data" - ) + network.name, + vpc_networks[0].name, + "Name of the VPC network should match with listVPC data" + ) if state: self.assertEqual( - vpc_networks[0].state, - state, + vpc_networks[0].state, + state, "VPC state should be '%s'" % state - ) + ) self.debug("VPC network validated - %s" % network.name) return @@ -1125,44 +1151,45 @@ class TestVPCNetworkRanges(cloudstackTestCase): # 1. Create a VPC with cidr - 10.1.1.1/16 # 2. Add network1 with cidr - 10.2.1.1/24 to this VPC # 3. Network creation should fail. - # 4. Repeat test for offering which has Netscaler as external LB provider + # 4. Repeat test for offering which has Netscaler as external LB + # provider - if (value == "network_offering_vpcns" and self.ns_configured == False): - self.skipTest('Netscaler not configured: skipping test') + if (value == "network_offering_vpcns" and not self.ns_configured): + self.skipTest('Netscaler not configured: skipping test') if (value == "network_offering"): - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering', - listall=True - ) + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering', + listall=True + ) else: - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering with Netscaler', - listall=True - ) - vpc_off=vpc_off_list[0] + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering with Netscaler', + listall=True + ) + vpc_off = vpc_off_list[0] self.debug("Creating a VPC with offering: %s" % vpc_off.id) self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.debug("Creating network offering") self.network_offering = NetworkOffering.create( - self.apiclient, - self.services[value], - conservemode=False - ) + self.apiclient, + self.services[value], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) @@ -1171,15 +1198,15 @@ class TestVPCNetworkRanges(cloudstackTestCase): self.debug("Creating network outside of the VPC's network") with self.assertRaises(Exception): Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.2.1.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.2.1.1', + vpcid=vpc.id + ) self.debug( "Network creation failed as network cidr range is outside of vpc") return @@ -1193,13 +1220,14 @@ class TestVPCNetworkRanges(cloudstackTestCase): # 1. Create a VPC with cidr - 10.1.1.1/16 # 2. Add network1 with cidr - 10.2.1.1/24 to this VPC # 3. Network creation should fail. - # 4. Repeat test for offering which has Netscaler as external LB provider + # 4. Repeat test for offering which has Netscaler as external LB + # provider self.debug("Creating a VPC offering") vpc_off = VpcOffering.create( - self.apiclient, - self.services["vpc_offering"] - ) + self.apiclient, + self.services["vpc_offering"] + ) self.cleanup.append(vpc_off) self.validate_vpc_offering(vpc_off) @@ -1210,22 +1238,22 @@ class TestVPCNetworkRanges(cloudstackTestCase): self.debug("creating a VPC network with cidr: 10.1.1.1/16") self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.debug("Creating network offering") self.network_offering = NetworkOffering.create( - self.apiclient, - self.services["network_offering"], - conservemode=False - ) + self.apiclient, + self.services["network_offering"], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) @@ -1234,15 +1262,15 @@ class TestVPCNetworkRanges(cloudstackTestCase): self.debug("Creating network outside of the VPC's network") with self.assertRaises(Exception): Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.2.1.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.2.1.1', + vpcid=vpc.id + ) self.debug( "Network creation failed as network cidr range is outside of vpc") return @@ -1257,45 +1285,46 @@ class TestVPCNetworkRanges(cloudstackTestCase): # 1. Create a VPC with cidr - 10.1.1.1/16 # 2. Add network1 with cidr - 10.1.1.1/8 to this VPC # 3. Network creation should fail. - # 4. Repeat test for offering which has Netscaler as external LB provider + # 4. Repeat test for offering which has Netscaler as external LB + # provider - if (value == "network_offering_vpcns" and self.ns_configured == False): - self.skipTest('Netscaler not configured: skipping test') + if (value == "network_offering_vpcns" and not self.ns_configured): + self.skipTest('Netscaler not configured: skipping test') if (value == "network_offering"): - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering', - listall=True - ) + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering', + listall=True + ) else: - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering with Netscaler', - listall=True - ) - vpc_off=vpc_off_list[0] + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering with Netscaler', + listall=True + ) + vpc_off = vpc_off_list[0] self.debug("Creating a VPC with offering: %s" % vpc_off.id) self.debug("creating a VPC network with cidr: 10.1.1.1/16") self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.debug("Creating network offering") self.network_offering = NetworkOffering.create( - self.apiclient, - self.services[value], - conservemode=False - ) + self.apiclient, + self.services[value], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) @@ -1307,15 +1336,15 @@ class TestVPCNetworkRanges(cloudstackTestCase): # cidr = 10.1.1.1/8 -> netmask = 255.0.0.0, gateway = 10.1.1.1 self.services["network"]["netmask"] = '255.0.0.0' Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.1.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.1.1', + vpcid=vpc.id + ) self.debug( "Network creation failed as network cidr range is inside of vpc") return @@ -1332,104 +1361,105 @@ class TestVPCNetworkRanges(cloudstackTestCase): # 3. Add network2 with cidr - 10.1.1.1/24 to this VPC # 4. Add network3 with cidr - 10.1.1.1/26 to this VPC # 5. Network creation in step 3 & 4 should fail. - # 6. Repeat test for offering which has Netscaler as external LB provider + # 6. Repeat test for offering which has Netscaler as external LB + # provider self.services = Services().services - if (value == "network_offering_vpcns" and self.ns_configured == False): - self.skipTest('Netscaler not configured: skipping test') + if (value == "network_offering_vpcns" and not self.ns_configured): + self.skipTest('Netscaler not configured: skipping test') if (value == "network_offering"): - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering', - listall=True - ) + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering', + listall=True + ) else: - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering with Netscaler', - listall=True - ) - vpc_off=vpc_off_list[0] + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering with Netscaler', + listall=True + ) + vpc_off = vpc_off_list[0] self.debug("Creating a VPC with offering: %s" % vpc_off.id) self.debug("creating a VPC network with cidr: 10.1.1.1/16") self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.debug("Creating network offering") self.network_offering = NetworkOffering.create( - self.apiclient, - self.services[value], - conservemode=False - ) + self.apiclient, + self.services[value], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) # Creating network using the network offering created self.debug("Creating network with network offering: %s" % - self.network_offering.id) + self.network_offering.id) network = Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.1.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.1.1', + vpcid=vpc.id + ) self.debug("Created network with ID: %s" % network.id) self.debug( "Verifying list network response to check if network created?") networks = Network.list( - self.apiclient, - id=network.id, - listall=True - ) + self.apiclient, + id=network.id, + listall=True + ) self.assertEqual( - isinstance(networks, list), - True, - "List networks should return a valid response" - ) + isinstance(networks, list), + True, + "List networks should return a valid response" + ) nw = networks[0] self.assertEqual( nw.networkofferingid, self.network_offering.id, "Network should be created from network offering - %s" % - self.network_offering.id - ) + self.network_offering.id + ) self.assertEqual( - nw.vpcid, - vpc.id, - "Network should be created in VPC: %s" % vpc.name - ) + nw.vpcid, + vpc.id, + "Network should be created in VPC: %s" % vpc.name + ) # Creating network using the network offering created self.debug( "Creating network with same network range as of previous network") with self.assertRaises(Exception): Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.1.1', - vpcid=vpc.id - ) - self.debug("Network creation as network range 10.1.1.1/24 is same" + \ + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.1.1', + vpcid=vpc.id + ) + self.debug("Network creation as network range 10.1.1.1/24 is same" + "as that of existing network") self.debug("Creating network having overlapping network ranges") @@ -1438,15 +1468,15 @@ class TestVPCNetworkRanges(cloudstackTestCase): self.services["network"]["netmask"] = '255.255.255.192' Network.create( - self.apiclient, - self.services["network"], - accountid=self.account.name, - domainid=self.account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.1.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=self.account.name, + domainid=self.account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.1.1', + vpcid=vpc.id + ) self.debug( "Network creation failed as network range overlaps each other") return @@ -1462,58 +1492,59 @@ class TestVPCNetworkRanges(cloudstackTestCase): # 2. Add network1 with cidr - 10.1.1.1/24 to this VPC # 3. Create another account # 4. Create network using this account - Network creation should fail - # 5. Repeat test for offering which has Netscaler as external LB provider + # 5. Repeat test for offering which has Netscaler as external LB + # provider - if (value == "network_offering_vpcns" and self.ns_configured == False): - self.skipTest('Netscaler not configured: skipping test') + if (value == "network_offering_vpcns" and not self.ns_configured): + self.skipTest('Netscaler not configured: skipping test') if (value == "network_offering"): - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering', - listall=True - ) + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering', + listall=True + ) else: - vpc_off_list=VpcOffering.list( - self.apiclient, - name='Default VPC offering with Netscaler', - listall=True - ) - vpc_off=vpc_off_list[0] + vpc_off_list = VpcOffering.list( + self.apiclient, + name='Default VPC offering with Netscaler', + listall=True + ) + vpc_off = vpc_off_list[0] self.debug("Creating a VPC with offering: %s" % vpc_off.id) self.debug("creating a VPC network with cidr: 10.1.1.1/16") self.services["vpc"]["cidr"] = '10.1.1.1/16' vpc = VPC.create( - self.apiclient, - self.services["vpc"], - vpcofferingid=vpc_off.id, - zoneid=self.zone.id, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + self.services["vpc"], + vpcofferingid=vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid + ) self.validate_vpc_network(vpc) self.debug("Creating network offering") self.network_offering = NetworkOffering.create( - self.apiclient, - self.services[value], - conservemode=False - ) + self.apiclient, + self.services[value], + conservemode=False + ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.cleanup.append(self.network_offering) self.debug( "Creating the new account to create new network in VPC: %s" % - vpc.name) + vpc.name) account = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) + self.apiclient, + self.services["account"], + admin=True, + domainid=self.domain.id + ) self.cleanup.append(account) # Creating network using the network offering created @@ -1523,19 +1554,20 @@ class TestVPCNetworkRanges(cloudstackTestCase): # cidr = 10.1.1.1/8 -> netmask = 255.0.0.0, gateway = 10.1.1.1 self.services["network"]["netmask"] = '255.0.0.0' Network.create( - self.apiclient, - self.services["network"], - accountid=account.name, - domainid=account.domainid, - networkofferingid=self.network_offering.id, - zoneid=self.zone.id, - gateway='10.1.1.1', - vpcid=vpc.id - ) + self.apiclient, + self.services["network"], + accountid=account.name, + domainid=account.domainid, + networkofferingid=self.network_offering.id, + zoneid=self.zone.id, + gateway='10.1.1.1', + vpcid=vpc.id + ) self.debug( "Network creation failed as VPC belongs to different account") return + class TestVPCNetworkUpgrade(cloudstackTestCase): @classmethod @@ -1549,18 +1581,18 @@ class TestVPCNetworkUpgrade(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.services["virtual_machine"]["zoneid"] = cls.zone.id cls.services["virtual_machine"]["template"] = cls.template.id cls._cleanup = [] cls.service_offering = ServiceOffering.create( - cls.api_client, - cls.services["service_offering"] - ) + cls.api_client, + cls.services["service_offering"] + ) cls._cleanup.append(cls.service_offering) return @@ -1568,7 +1600,7 @@ class TestVPCNetworkUpgrade(cloudstackTestCase): @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) @@ -1578,11 +1610,11 @@ class TestVPCNetworkUpgrade(cloudstackTestCase): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.account = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) + self.apiclient, + self.services["account"], + admin=True, + domainid=self.domain.id + ) self.cleanup = [self.account, ] return @@ -1598,22 +1630,22 @@ class TestVPCNetworkUpgrade(cloudstackTestCase): self.debug("Check if the VPC offering is created successfully?") vpc_offs = VpcOffering.list( - self.apiclient, - id=vpc_offering.id - ) + self.apiclient, + id=vpc_offering.id + ) self.assertEqual( - isinstance(vpc_offs, list), - True, - "List VPC offerings should return a valid list" - ) + isinstance(vpc_offs, list), + True, + "List VPC offerings should return a valid list" + ) self.assertEqual( - vpc_offering.name, - vpc_offs[0].name, - "Name of the VPC offering should match with listVPCOff data" - ) + vpc_offering.name, + vpc_offs[0].name, + "Name of the VPC offering should match with listVPCOff data" + ) self.debug( - "VPC offering is created successfully - %s" % - vpc_offering.name) + "VPC offering is created successfully - %s" % + vpc_offering.name) return def validate_vpc_network(self, network, state=None): @@ -1621,31 +1653,32 @@ class TestVPCNetworkUpgrade(cloudstackTestCase): self.debug("Check if the VPC network is created successfully?") vpc_networks = VPC.list( - self.apiclient, - id=network.id - ) + self.apiclient, + id=network.id + ) self.assertEqual( - isinstance(vpc_networks, list), - True, - "List VPC network should return a valid list" - ) + isinstance(vpc_networks, list), + True, + "List VPC network should return a valid list" + ) self.assertEqual( - network.name, - vpc_networks[0].name, - "Name of the VPC network should match with listVPC data" - ) + network.name, + vpc_networks[0].name, + "Name of the VPC network should match with listVPC data" + ) if state: self.assertEqual( - vpc_networks[0].state, - state, + vpc_networks[0].state, + state, "VPC state should be '%s'" % state - ) + ) self.debug("VPC network validated - %s" % network.name) return @attr(tags=["advanced", "intervlan"], required_hardware="true") def test_01_network_services_upgrade(self): - """ Test update Network that is part of a VPC to a network offering that h