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 B8175106B1 for ; Thu, 26 Feb 2015 07:36:39 +0000 (UTC) Received: (qmail 42218 invoked by uid 500); 26 Feb 2015 07:36:39 -0000 Delivered-To: apmail-cloudstack-commits-archive@cloudstack.apache.org Received: (qmail 42027 invoked by uid 500); 26 Feb 2015 07:36:39 -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 42010 invoked by uid 99); 26 Feb 2015 07:36:39 -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; Thu, 26 Feb 2015 07:36:39 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id 3814BE03B4; Thu, 26 Feb 2015 07:36:39 +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: Thu, 26 Feb 2015 07:36:41 -0000 Message-Id: In-Reply-To: <16adbb363cf24d618dacc922e7316934@git.apache.org> References: <16adbb363cf24d618dacc922e7316934@git.apache.org> X-Mailer: ASF-Git Admin Mailer Subject: [3/4] git commit: updated refs/heads/master to f393199 http://git-wip-us.apache.org/repos/asf/cloudstack/blob/f3931993/test/integration/component/test_multiple_ips_per_nic.py ---------------------------------------------------------------------- diff --git a/test/integration/component/test_multiple_ips_per_nic.py b/test/integration/component/test_multiple_ips_per_nic.py index 58aac9f..536c39d 100644 --- a/test/integration/component/test_multiple_ips_per_nic.py +++ b/test/integration/component/test_multiple_ips_per_nic.py @@ -23,15 +23,34 @@ Design Document: https://cwiki.apache.org/confluence/display/CLOUDSTACK/Multiple+IP+address+per+NIC """ from marvin.cloudstackTestCase import cloudstackTestCase -from marvin.lib.utils import * -from marvin.lib.base import * -from marvin.lib.common import * - +from marvin.lib.utils import (cleanup_resources, + random_gen, + validateList) +from marvin.lib.base import (Account, + VirtualMachine, + PublicIPAddress, + NATRule, + StaticNATRule, + FireWallRule, + NIC, + Network, + VPC, + ServiceOffering, + VpcOffering, + Domain, + Router) +from marvin.lib.common import (get_domain, + get_template, + get_zone, + setSharedNetworkParams, + get_free_vlan, + createEnabledNetworkOffering) from nose.plugins.attrib import attr -from marvin.codes import PASS, ISOLATED_NETWORK, VPC_NETWORK, SHARED_NETWORK, FAIL +from marvin.codes import PASS, ISOLATED_NETWORK, VPC_NETWORK, SHARED_NETWORK, FAIL, FAILED from ddt import ddt, data import time + def createNetwork(self, networkType): """Create a network of given type (isolated/shared/isolated in VPC)""" @@ -39,27 +58,36 @@ def createNetwork(self, networkType): if networkType == ISOLATED_NETWORK: try: - network = Network.create(self.apiclient,self.services["isolated_network"], - networkofferingid=self.isolated_network_offering.id, - accountid=self.account.name,domainid=self.account.domainid, - zoneid=self.zone.id) + network = Network.create( + self.apiclient, + self.services["isolated_network"], + networkofferingid=self.isolated_network_offering.id, + accountid=self.account.name, + domainid=self.account.domainid, + zoneid=self.zone.id) except Exception as e: self.fail("Isolated network creation failed because: %s" % e) elif networkType == SHARED_NETWORK: physical_network, vlan = get_free_vlan(self.api_client, self.zone.id) - #create network using the shared network offering created + # create network using the shared network offering created self.services["shared_network"]["acltype"] = "domain" self.services["shared_network"]["vlan"] = vlan - self.services["shared_network"]["networkofferingid"] = self.shared_network_offering.id - self.services["shared_network"]["physicalnetworkid"] = physical_network.id + self.services["shared_network"][ + "networkofferingid"] = self.shared_network_offering.id + self.services["shared_network"][ + "physicalnetworkid"] = physical_network.id - self.services["shared_network"] = setSharedNetworkParams(self.services["shared_network"]) + self.services["shared_network"] = setSharedNetworkParams( + self.services["shared_network"]) try: - network = Network.create(self.api_client, self.services["shared_network"], - networkofferingid=self.shared_network_offering.id, zoneid=self.zone.id) + network = Network.create( + self.api_client, + self.services["shared_network"], + networkofferingid=self.shared_network_offering.id, + zoneid=self.zone.id) self.cleanup.append(network) except Exception as e: self.fail("Shared Network creation failed because: %s" % e) @@ -67,53 +95,96 @@ def createNetwork(self, networkType): elif networkType == VPC_NETWORK: self.services["vpc"]["cidr"] = "10.1.1.1/16" self.debug("creating a VPC network in the account: %s" % - self.account.name) - vpc = VPC.create(self.apiclient, self.services["vpc"], - vpcofferingid=self.vpc_off.id, zoneid=self.zone.id, - account=self.account.name, domainid=self.account.domainid) + self.account.name) + vpc = VPC.create( + self.apiclient, + self.services["vpc"], + vpcofferingid=self.vpc_off.id, + zoneid=self.zone.id, + account=self.account.name, + domainid=self.account.domainid) vpcs = VPC.list(self.apiclient, id=vpc.id) - self.assertEqual(validateList(vpcs)[0], PASS, "VPC list validation failed, vpc list is %s" % vpcs) - - network = Network.create(self.api_client,self.services["isolated_network"], - networkofferingid=self.isolated_network_offering_vpc.id, - accountid=self.account.name,domainid=self.account.domainid, - zoneid=self.zone.id, vpcid=vpc.id, gateway="10.1.1.1", netmask="255.255.255.0") + self.assertEqual( + validateList(vpcs)[0], + PASS, + "VPC list validation failed, vpc list is %s" % + vpcs) + + network = Network.create( + self.api_client, + self.services["isolated_network"], + networkofferingid=self.isolated_network_offering_vpc.id, + accountid=self.account.name, + domainid=self.account.domainid, + zoneid=self.zone.id, + vpcid=vpc.id, + gateway="10.1.1.1", + netmask="255.255.255.0") return network + def CreateEnabledNetworkOffering(apiclient, networkServices): """Create network offering of given services and enable it""" result = createEnabledNetworkOffering(apiclient, networkServices) - assert result[0] == PASS, "Network offering creation/enabling failed due to %s" % result[2] + assert result[ + 0] == PASS, "Network offering creation/enabling failed due to %s" % result[2] return result[1] -def createNetworkRules(self, virtual_machine, network, vmguestip, networktype, ruletype): + +def createNetworkRules( + self, + virtual_machine, + network, + vmguestip, + networktype, + ruletype): """ Acquire public ip in the given network, open firewall if required and create NAT rule for the public ip to the given guest vm ip address""" try: - public_ip = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if networktype == VPC_NETWORK else None) + public_ip = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if networktype == VPC_NETWORK else None) if networktype != VPC_NETWORK: - FireWallRule.create(self.apiclient,ipaddressid=public_ip.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) + FireWallRule.create( + self.apiclient, + ipaddressid=public_ip.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) if ruletype == "nat": - NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=public_ip.ipaddress.id, - networkid=network.id, vmguestip = vmguestip) + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=public_ip.ipaddress.id, + networkid=network.id, + vmguestip=vmguestip) elif ruletype == "staticnat": - StaticNATRule.enable(self.apiclient, public_ip.ipaddress.id, virtual_machine.id, network.id, vmguestip=vmguestip) - except Exception: + StaticNATRule.enable( + self.apiclient, + public_ip.ipaddress.id, + virtual_machine.id, + network.id, + vmguestip=vmguestip) + except Exception as e: + self.debug("Exception occured while creating network rules: %s" % e) return FAIL - return PASS + @ddt class TestBasicOperations(cloudstackTestCase): + """Test Basic operations (add/remove/list) IP to/from NIC """ @@ -129,34 +200,42 @@ class TestBasicOperations(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.api_client, + cls.services["service_offering"] + ) cls._cleanup = [cls.service_offering] - cls.services["shared_network_offering_all_services"]["specifyVlan"] = "True" - cls.services["shared_network_offering_all_services"]["specifyIpRanges"] = "True" - - cls.shared_network_offering = CreateEnabledNetworkOffering(cls.api_client, - cls.services["shared_network_offering_all_services"]) + cls.services["shared_network_offering_all_services"][ + "specifyVlan"] = "True" + cls.services["shared_network_offering_all_services"][ + "specifyIpRanges"] = "True" + + cls.shared_network_offering = CreateEnabledNetworkOffering( + cls.api_client, + cls.services["shared_network_offering_all_services"]) cls._cleanup.append(cls.shared_network_offering) cls.mode = cls.zone.networktype - cls.isolated_network_offering = CreateEnabledNetworkOffering(cls.api_client, - cls.services["isolated_network_offering"]) + cls.isolated_network_offering = CreateEnabledNetworkOffering( + cls.api_client, + cls.services["isolated_network_offering"]) cls._cleanup.append(cls.isolated_network_offering) - cls.isolated_network_offering_vpc = CreateEnabledNetworkOffering(cls.api_client, - cls.services["nw_offering_isolated_vpc"]) + cls.isolated_network_offering_vpc = CreateEnabledNetworkOffering( + cls.api_client, + cls.services["nw_offering_isolated_vpc"]) cls._cleanup.append(cls.isolated_network_offering_vpc) - cls.vpc_off = VpcOffering.create(cls.api_client, cls.services["vpc_offering"]) + cls.vpc_off = VpcOffering.create( + cls.api_client, + cls.services["vpc_offering"]) cls.vpc_off.update(cls.api_client, state='Enabled') cls._cleanup.append(cls.vpc_off) return @@ -173,7 +252,7 @@ class TestBasicOperations(cloudstackTestCase): def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() - self.cleanup = [ ] + self.cleanup = [] return def tearDown(self): @@ -188,14 +267,24 @@ class TestBasicOperations(cloudstackTestCase): def VerifyStaticNatForPublicIp(self, ipaddressid, natrulestatus): """ List public IP and verify that NAT rule status for the IP is as desired """ - publiciplist = PublicIPAddress.list(self.apiclient, id=ipaddressid, listall=True) - self.assertEqual(validateList(publiciplist)[0], PASS, "Public IP list validation failed") - self.assertEqual(publiciplist[0].isstaticnat, natrulestatus, "isstaticnat should be %s, it is %s" % - (natrulestatus, publiciplist[0].isstaticnat)) + publiciplist = PublicIPAddress.list( + self.apiclient, + id=ipaddressid, + listall=True) + self.assertEqual( + validateList(publiciplist)[0], + PASS, + "Public IP list validation failed") + self.assertEqual( + publiciplist[0].isstaticnat, + natrulestatus, + "isstaticnat should be %s, it is %s" % + (natrulestatus, + publiciplist[0].isstaticnat)) return - @data(ISOLATED_NETWORK, SHARED_NETWORK, VPC_NETWORK) + @data(SHARED_NETWORK) @attr(tags=["advanced"]) def test_add_ip_to_nic(self, value): """ Add secondary IP to NIC of a VM""" @@ -206,7 +295,8 @@ class TestBasicOperations(cloudstackTestCase): # 3. Add secondary IP to the default nic of VM # 4. Try to add the same IP again # 5. Try to add secondary IP providing wrong virtual machine id - # 6. Try to add secondary IP with correct virtual machine id but wrong IP address + # 6. Try to add secondary IP with correct virtual machine id but wrong + # IP address # Validations: # 1. Step 3 should succeed @@ -214,41 +304,60 @@ class TestBasicOperations(cloudstackTestCase): # 3. Step 5 should should fail # 4. Step 6 should fail - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) try: - NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id, ipaddress=ipaddress_1.ipaddress) - self.debug("Adding already added secondary IP %s to NIC of vm %s succeeded, should have failed" % - (ipaddress_1.ipaddress, virtual_machine.id)) + NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id, + ipaddress=ipaddress_1.ipaddress) + self.debug( + "Adding already added secondary IP %s to NIC of vm %s succeeded, should have failed" % + (ipaddress_1.ipaddress, virtual_machine.id)) except Exception as e: - self.debug("Failed while adding already added secondary IP to NIC of vm %s" % virtual_machine.id) + self.debug( + "Failed while adding already added secondary IP to NIC of vm %s" % + virtual_machine.id) try: - NIC.addIp(self.apiclient, id=(virtual_machine.nic[0].id + random_gen())) - self.fail("Adding secondary IP with wrong NIC id succeded, it shoud have failed") + NIC.addIp( + self.apiclient, id=( + virtual_machine.nic[0].id + random_gen())) + self.fail( + "Adding secondary IP with wrong NIC id succeded, it shoud have failed") except Exception as e: self.debug("Failed while adding secondary IP to wrong NIC") try: - NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id, ipaddress = "255.255.255.300") - self.fail("Adding secondary IP with wrong ipaddress succeded, it should have failed") + NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id, + ipaddress="255.255.255.300") + self.fail( + "Adding secondary IP with wrong ipaddress succeded, it should have failed") except Exception as e: - self.debug("Failed while adding wrong secondary IP to NIC of VM %s" % virtual_machine.id) + self.debug( + "Failed while adding wrong secondary IP to NIC of VM %s: %s" % + (virtual_machine.id, e)) return @data(ISOLATED_NETWORK, SHARED_NETWORK, VPC_NETWORK) @@ -267,33 +376,39 @@ class TestBasicOperations(cloudstackTestCase): # 1. Step 4 should succeed # 2. Step 5 should fail - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) - - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.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.account.domainid) - try: - NIC.removeIp(self.apiclient, ipaddressid=ipaddress_1.id) - except Exception as e: - self.fail("Removing seondary IP %s from NIC failed as expected with Exception %s" % (ipaddress_1.id,e)) + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + NIC.removeIp(self.apiclient, ipaddressid=ipaddress_1.id) try: - NIC.removeIp(self.apiclient, ipaddressid=(ipaddress_1.id + random_gen())) + NIC.removeIp( + self.apiclient, + ipaddressid=( + ipaddress_1.id + + random_gen())) self.fail("Removing invalid IP address, it should have failed") except Exception as e: - self.debug("Removing invalid IP failed as expected with Exception %s" % e) + self.debug( + "Removing invalid IP failed as expected with Exception %s" % + e) return @attr(tags=["advanced"]) @@ -308,9 +423,12 @@ class TestBasicOperations(cloudstackTestCase): try: NIC.removeIp(self.apiclient, ipaddressid="") - self.fail("Removing IP address without passing IP succeeded, it should have failed") + self.fail( + "Removing IP address without passing IP succeeded, it should have failed") except Exception as e: - self.debug("Removing IP from NIC without passing ipaddressid failed as expected with Exception %s" % e) + self.debug( + "Removing IP from NIC without passing ipaddressid failed as expected with Exception %s" % + e) return @data(ISOLATED_NETWORK, SHARED_NETWORK, VPC_NETWORK) @@ -327,7 +445,8 @@ class TestBasicOperations(cloudstackTestCase): # 6. Try to list secondary IPs by passing correct vm id and its nic id # 7. Try to list secondary IPs by passing incorrect vm id and correct nic id # 8. Try to list secondary IPs by passing correct vm id and incorrect nic id - # 9. Try to list secondary IPs by passing incorrect vm id and incorrect nic id + # 9. Try to list secondary IPs by passing incorrect vm id and incorrect + # nic id # Validations: # 1. Step 4 should fail @@ -337,58 +456,89 @@ class TestBasicOperations(cloudstackTestCase): # 5. Step 8 should fail # 6. Step 9 should fail - - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) - try: - NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) + NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) try: nics = NIC.list(self.apiclient) - self.fail("Listing NICs without passign VM id succeeded, it should have failed, list is %s" % nics) + self.fail( + "Listing NICs without passign VM id succeeded, it should have failed, list is %s" % + nics) except Exception as e: - self.debug("Listing NICs without passing virtual machine id failed as expected") + self.debug( + "Listing NICs without passing virtual machine id failed as expected") try: NIC.list(self.apiclient, virtualmachineid=virtual_machine.id) except Exception as e: - self.fail("Listing NICs for virtual machine %s failed with Exception %s" % (virtual_machine.id, e)) + self.fail( + "Listing NICs for virtual machine %s failed with Exception %s" % + (virtual_machine.id, e)) - try: - NIC.list(self.apiclient, virtualmachineid=virtual_machine.id, nicid=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Listing NICs for virtual machine %s and nic id %s failed with Exception %s" % - (virtual_machine.id, virtual_machine.nic[0].id, e)) + NIC.list( + self.apiclient, + virtualmachineid=virtual_machine.id, + nicid=virtual_machine.nic[0].id) try: - nics = NIC.list(self.apiclient, virtualmachineid=(virtual_machine.id + random_gen()), nicid=virtual_machine.nic[0].id) - self.fail("Listing NICs with wrong virtual machine id and right nic id succeeded, should have failed") + nics = NIC.list( + self.apiclient, + virtualmachineid=( + virtual_machine.id + + random_gen()), + nicid=virtual_machine.nic[0].id) + self.fail( + "Listing NICs with wrong virtual machine id and right nic id succeeded, should have failed") except Exception as e: - self.debug("Listing NICs with wrong virtual machine id and right nic failed as expected with Exception %s" % e) + self.debug( + "Listing NICs with wrong virtual machine id and right nic failed as expected with Exception %s" % + e) try: - nics = NIC.list(self.apiclient, virtualmachineid=virtual_machine.id, nicid=(virtual_machine.nic[0].id + random_gen())) - self.fail("Listing NICs with correct virtual machine id but wrong nic id succeeded, should have failed") + nics = NIC.list( + self.apiclient, + virtualmachineid=virtual_machine.id, + nicid=( + virtual_machine.nic[0].id + + random_gen())) + self.fail( + "Listing NICs with correct virtual machine id but wrong nic id succeeded, should have failed") except Exception as e: - self.debug("Listing NICs with correct virtual machine id but wrong nic id failed as expected with Exception %s" % e) + self.debug( + "Listing NICs with correct virtual machine id but wrong nic id failed as expected with Exception %s" % + e) try: - nics = NIC.list(self.apiclient, virtualmachineid=(virtual_machine.id+random_gen()), nicid=(virtual_machine.nic[0].id + random_gen())) - self.fail("Listing NICs with wrong virtual machine id and wrong nic id succeeded, should have failed") + nics = NIC.list( + self.apiclient, + virtualmachineid=( + virtual_machine.id + + random_gen()), + nicid=( + virtual_machine.nic[0].id + + random_gen())) + self.fail( + "Listing NICs with wrong virtual machine id and wrong nic id succeeded, should have failed") except Exception as e: - self.debug("Listing NICs with wrong virtual machine id and wrong nic id failed as expected with Exception %s" % e) + self.debug( + "Listing NICs with wrong virtual machine id and wrong nic id failed as expected with Exception %s" % + e) return @@ -409,49 +559,65 @@ class TestBasicOperations(cloudstackTestCase): # Validations: # 1. All the operations should be successful - child_domain = Domain.create(self.apiclient,services=self.services["domain"], - parentdomainid=self.domain.id) + child_domain = Domain.create( + self.apiclient, + services=self.services["domain"], + parentdomainid=self.domain.id) - self.account = Account.create(self.apiclient,self.services["account"],domainid=child_domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=child_domain.id) self.cleanup.append(self.account) self.cleanup.append(child_domain) - apiclient = self.testClient.getUserApiClient(UserName=self.account.name, DomainName=self.account.domain) + apiclient = self.testClient.getUserApiClient( + UserName=self.account.name, + DomainName=self.account.domain) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) - try: - ipaddress_1 = NIC.addIp(apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s: %s" % (virtual_machine.id, e)) + ipaddress_1 = NIC.addIp(apiclient, id=virtual_machine.nic[0].id) try: NIC.list(apiclient, virtualmachineid=virtual_machine.id) except Exception as e: - self.fail("Listing NICs for virtual machine %s failed with Exception %s" % (virtual_machine.id, e)) + self.fail( + "Listing NICs for virtual machine %s failed with Exception %s" % + (virtual_machine.id, e)) try: - NIC.list(apiclient, virtualmachineid=virtual_machine.id, nicid=virtual_machine.nic[0].id) + NIC.list( + apiclient, + virtualmachineid=virtual_machine.id, + nicid=virtual_machine.nic[0].id) except Exception as e: - self.fail("Listing NICs for virtual machine %s and nic id %s failed with Exception %s" % - (virtual_machine.id, virtual_machine.nic[0].id, e)) + self.fail( + "Listing NICs for virtual machine %s and nic id %s failed with Exception %s" % + (virtual_machine.id, virtual_machine.nic[0].id, e)) try: NIC.removeIp(apiclient, ipaddressid=ipaddress_1.id) except Exception as e: - self.fail("Removing seondary IP %s from NIC failed as expected with Exception %s" % (ipaddress_1.id,e)) + self.fail( + "Removing seondary IP %s from NIC failed as expected with Exception %s" % + (ipaddress_1.id, e)) return + @ddt class TestNetworkRules(cloudstackTestCase): + """Test PF/NAT/static nat rules with the secondary IPs """ @@ -467,32 +633,42 @@ class TestNetworkRules(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.api_client, + cls.services["service_offering"] + ) cls._cleanup = [cls.service_offering] - cls.services["shared_network_offering_all_services"]["specifyVlan"] = "True" - cls.services["shared_network_offering_all_services"]["specifyIpRanges"] = "True" - - cls.shared_network_offering = CreateEnabledNetworkOffering(cls.api_client, - cls.services["shared_network_offering_all_services"]) + cls.services["shared_network_offering_all_services"][ + "specifyVlan"] = "True" + cls.services["shared_network_offering_all_services"][ + "specifyIpRanges"] = "True" + + cls.shared_network_offering = CreateEnabledNetworkOffering( + cls.api_client, + cls.services["shared_network_offering_all_services"]) cls._cleanup.append(cls.shared_network_offering) cls.mode = cls.zone.networktype - cls.isolated_network_offering = CreateEnabledNetworkOffering(cls.api_client, cls.services["isolated_network_offering"]) + cls.isolated_network_offering = CreateEnabledNetworkOffering( + cls.api_client, + cls.services["isolated_network_offering"]) cls._cleanup.append(cls.isolated_network_offering) - cls.isolated_network_offering_vpc = CreateEnabledNetworkOffering(cls.api_client, cls.services["nw_offering_isolated_vpc"]) + cls.isolated_network_offering_vpc = CreateEnabledNetworkOffering( + cls.api_client, + cls.services["nw_offering_isolated_vpc"]) cls._cleanup.append(cls.isolated_network_offering_vpc) - cls.vpc_off = VpcOffering.create(cls.api_client, cls.services["vpc_offering"]) + cls.vpc_off = VpcOffering.create( + cls.api_client, + cls.services["vpc_offering"]) cls.vpc_off.update(cls.api_client, state='Enabled') cls._cleanup.append(cls.vpc_off) return @@ -509,7 +685,7 @@ class TestNetworkRules(cloudstackTestCase): def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() - self.cleanup = [ ] + self.cleanup = [] return def tearDown(self): @@ -524,10 +700,20 @@ class TestNetworkRules(cloudstackTestCase): def VerifyStaticNatForPublicIp(self, ipaddressid, natrulestatus): """ List public IP and verify that NAT rule status for the IP is as desired """ - publiciplist = PublicIPAddress.list(self.apiclient, id=ipaddressid, listall=True) - self.assertEqual(validateList(publiciplist)[0], PASS, "Public IP list validation failed") - self.assertEqual(publiciplist[0].isstaticnat, natrulestatus, "isstaticnat should be %s, it is %s" % - (natrulestatus, publiciplist[0].isstaticnat)) + publiciplist = PublicIPAddress.list( + self.apiclient, + id=ipaddressid, + listall=True) + self.assertEqual( + validateList(publiciplist)[0], + PASS, + "Public IP list validation failed") + self.assertEqual( + publiciplist[0].isstaticnat, + natrulestatus, + "isstaticnat should be %s, it is %s" % + (natrulestatus, + publiciplist[0].isstaticnat)) return @@ -552,36 +738,71 @@ class TestNetworkRules(cloudstackTestCase): # 3. Step 6 should succeed # 4. Step 7 should fail - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) - - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) - - try: - ipaddress_2 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) - - self.assertEqual(createNetworkRules(self, virtual_machine, network, ipaddress_1.ipaddress, value, ruletype="nat"), - PASS, "Failure in creating NAT rule") - self.assertEqual(createNetworkRules(self, virtual_machine, network, ipaddress_1.ipaddress, value, ruletype="nat"), - PASS, "Failure in creating NAT rule") - self.assertEqual(createNetworkRules(self, virtual_machine, network, ipaddress_2.ipaddress, value, ruletype="nat"), - PASS, "Failure in creating NAT rule") - self.assertEqual(createNetworkRules(self, virtual_machine, network, "255.255.255.300", value, ruletype="nat"), - FAIL, "Failure in NAT rule creation") + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) + + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + + ipaddress_2 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + + self.assertEqual( + createNetworkRules( + self, + virtual_machine, + network, + ipaddress_1.ipaddress, + value, + ruletype="nat"), + PASS, + "Failure in creating NAT rule") + self.assertEqual( + createNetworkRules( + self, + virtual_machine, + network, + ipaddress_1.ipaddress, + value, + ruletype="nat"), + PASS, + "Failure in creating NAT rule") + self.assertEqual( + createNetworkRules( + self, + virtual_machine, + network, + ipaddress_2.ipaddress, + value, + ruletype="nat"), + PASS, + "Failure in creating NAT rule") + self.assertEqual( + createNetworkRules( + self, + virtual_machine, + network, + "255.255.255.300", + value, + ruletype="nat"), + FAIL, + "Failure in NAT rule creation") return @data(ISOLATED_NETWORK, SHARED_NETWORK, VPC_NETWORK) @@ -602,61 +823,81 @@ class TestNetworkRules(cloudstackTestCase): # 1. Step 5 should fail # 2. Step 6 should succeed - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) firewallrule = None - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) - - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) - - public_ip = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) + + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + + public_ip = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) if value != VPC_NETWORK: - firewallrule = FireWallRule.create(self.apiclient,ipaddressid=public_ip.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) + firewallrule = FireWallRule.create( + self.apiclient, + ipaddressid=public_ip.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) # Create NAT rule - natrule = NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=public_ip.ipaddress.id, - networkid=network.id, vmguestip = ipaddress_1.ipaddress) + natrule = NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=public_ip.ipaddress.id, + networkid=network.id, + vmguestip=ipaddress_1.ipaddress) try: NIC.removeIp(self.apiclient, ipaddressid=ipaddress_1.id) - self.fail("Removing secondary IP succeeded while it had active NAT rule on it, should have failed") + self.fail( + "Removing secondary IP succeeded while it had active NAT rule on it, should have failed") except Exception as e: - self.debug("Removing secondary IP with active NAT rule failed as expected") + self.debug( + "Removing secondary IP with active NAT rule failed as expected") if firewallrule: try: firewallrule.delete(self.apiclient) except Exception as e: - self.fail("Exception while deleting firewall rule %s: %s" % (firewallrule.id, e)) + self.fail( + "Exception while deleting firewall rule %s: %s" % + (firewallrule.id, e)) - try: - natrule.delete(self.apiclient) - except Exception as e: - self.fail("Exception while deleting nat rule %s: %s" % (natrule.id, e)) + natrule.delete(self.apiclient) return @data(ISOLATED_NETWORK, SHARED_NETWORK, VPC_NETWORK) @attr(tags=["advanced"]) - def test_disassociate_ip_mapped_to_secondary_ip_through_PF_rule(self, value): + def test_disassociate_ip_mapped_to_secondary_ip_through_PF_rule( + self, + value): """ Add secondary IP to NIC of a VM""" - ## Steps: + # Steps: # 1. Create Account and create network in it (isoalted/ shared/ vpc) # 2. Deploy a VM in this network and account # 3. Add secondary IP to the default nic of VM @@ -667,41 +908,55 @@ class TestNetworkRules(cloudstackTestCase): # Validations: # 1. Step 5 should succeed - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) - - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) - - public_ip = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) + + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + + public_ip = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) if value != VPC_NETWORK: - FireWallRule.create(self.apiclient,ipaddressid=public_ip.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) + FireWallRule.create( + self.apiclient, + ipaddressid=public_ip.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) # Create NAT rule - natrule = NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=public_ip.ipaddress.id, - networkid=network.id, vmguestip = ipaddress_1.ipaddress) + NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=public_ip.ipaddress.id, + networkid=network.id, + vmguestip=ipaddress_1.ipaddress) - try: - public_ip.delete(self.apiclient) - except Exception as e: - self.fail("Exception while deleting nat rule %s: %s" % (natrule.id, e)) + public_ip.delete(self.apiclient) return @data(ISOLATED_NETWORK, SHARED_NETWORK, VPC_NETWORK) @@ -727,42 +982,80 @@ class TestNetworkRules(cloudstackTestCase): # 4. Step 7 should fail # 5. Step 8 should succeed - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) - - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) - - try: - ipaddress_2 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) - - self.assertEqual(createNetworkRules(self, virtual_machine, network, ipaddress_1.ipaddress, value, ruletype="staticnat"), - PASS, "Failure in creating NAT rule") - self.assertEqual(createNetworkRules(self, virtual_machine, network, ipaddress_1.ipaddress, value, ruletype="staticnat"), - FAIL, "Failure in creating NAT rule") - self.assertEqual(createNetworkRules(self, virtual_machine, network, ipaddress_2.ipaddress, value, ruletype="staticnat"), - PASS, "Failure in creating NAT rule") - self.assertEqual(createNetworkRules(self, virtual_machine, network, "255.255.255.300", value, ruletype="staticnat"), - FAIL, "Failure in NAT rule creation") + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) + + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + + ipaddress_2 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + + self.assertEqual( + createNetworkRules( + self, + virtual_machine, + network, + ipaddress_1.ipaddress, + value, + ruletype="staticnat"), + PASS, + "Failure in creating NAT rule") + self.assertEqual( + createNetworkRules( + self, + virtual_machine, + network, + ipaddress_1.ipaddress, + value, + ruletype="staticnat"), + FAIL, + "Failure in creating NAT rule") + self.assertEqual( + createNetworkRules( + self, + virtual_machine, + network, + ipaddress_2.ipaddress, + value, + ruletype="staticnat"), + PASS, + "Failure in creating NAT rule") + self.assertEqual( + createNetworkRules( + self, + virtual_machine, + network, + "255.255.255.300", + value, + ruletype="staticnat"), + FAIL, + "Failure in NAT rule creation") try: NIC.removeIp(self.apiclient, ipaddress_1.id) - self.fail("Ip address should not get removed when active static NAT rule is defined for it") + self.fail( + "Ip address should not get removed when active static NAT rule is defined for it") except Exception as e: - self.debug("Exception while removing secondary ip address as expected because static nat rule is present for it") + self.debug( + "Exception while removing secondary ip address as expected\ + because static nat rule is present for it: %s" % e) return @data(ISOLATED_NETWORK, SHARED_NETWORK, VPC_NETWORK) @@ -779,36 +1072,54 @@ class TestNetworkRules(cloudstackTestCase): # 5. Disable the static nat rule and enable it again # Validations: - # 1. Verify step 5 by listing seconday IP and checking the appropriate flag + # 1. Verify step 5 by listing seconday IP and checking the appropriate + # flag - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) - - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) - - public_ip = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) + + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + + public_ip = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) if value != VPC_NETWORK: - FireWallRule.create(self.apiclient,ipaddressid=public_ip.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - - StaticNATRule.enable(self.apiclient, public_ip.ipaddress.id, virtual_machine.id, - network.id, vmguestip=ipaddress_1.ipaddress) + FireWallRule.create( + self.apiclient, + ipaddressid=public_ip.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + + StaticNATRule.enable( + self.apiclient, + public_ip.ipaddress.id, + virtual_machine.id, + network.id, + vmguestip=ipaddress_1.ipaddress) self.VerifyStaticNatForPublicIp(public_ip.ipaddress.id, True) @@ -817,16 +1128,22 @@ class TestNetworkRules(cloudstackTestCase): self.VerifyStaticNatForPublicIp(public_ip.ipaddress.id, False) - StaticNATRule.enable(self.apiclient, public_ip.ipaddress.id, virtual_machine.id, - network.id, vmguestip=ipaddress_1.ipaddress) + StaticNATRule.enable( + self.apiclient, + public_ip.ipaddress.id, + virtual_machine.id, + network.id, + vmguestip=ipaddress_1.ipaddress) self.VerifyStaticNatForPublicIp(public_ip.ipaddress.id, True) public_ip.delete(self.apiclient) return + @ddt class TestVmNetworkOperations(cloudstackTestCase): + """Test VM and Network operations with network rules created on secondary IP """ @@ -842,31 +1159,41 @@ class TestVmNetworkOperations(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.api_client, + cls.services["service_offering"] + ) cls._cleanup = [cls.service_offering] - cls.services["shared_network_offering_all_services"]["specifyVlan"] = "True" - cls.services["shared_network_offering_all_services"]["specifyIpRanges"] = "True" - - cls.shared_network_offering = CreateEnabledNetworkOffering(cls.api_client, - cls.services["shared_network_offering_all_services"]) + cls.services["shared_network_offering_all_services"][ + "specifyVlan"] = "True" + cls.services["shared_network_offering_all_services"][ + "specifyIpRanges"] = "True" + + cls.shared_network_offering = CreateEnabledNetworkOffering( + cls.api_client, + cls.services["shared_network_offering_all_services"]) cls._cleanup.append(cls.shared_network_offering) cls.mode = cls.zone.networktype - cls.isolated_network_offering = CreateEnabledNetworkOffering(cls.api_client, cls.services["isolated_network_offering"]) + cls.isolated_network_offering = CreateEnabledNetworkOffering( + cls.api_client, + cls.services["isolated_network_offering"]) cls._cleanup.append(cls.isolated_network_offering) - cls.isolated_network_offering_vpc = CreateEnabledNetworkOffering(cls.api_client, cls.services["nw_offering_isolated_vpc"]) + cls.isolated_network_offering_vpc = CreateEnabledNetworkOffering( + cls.api_client, + cls.services["nw_offering_isolated_vpc"]) cls._cleanup.append(cls.isolated_network_offering_vpc) - cls.vpc_off = VpcOffering.create(cls.api_client, cls.services["vpc_offering"]) + cls.vpc_off = VpcOffering.create( + cls.api_client, + cls.services["vpc_offering"]) cls.vpc_off.update(cls.api_client, state='Enabled') cls._cleanup.append(cls.vpc_off) return @@ -883,7 +1210,7 @@ class TestVmNetworkOperations(cloudstackTestCase): def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() - self.cleanup = [ ] + self.cleanup = [] return def tearDown(self): @@ -897,10 +1224,20 @@ class TestVmNetworkOperations(cloudstackTestCase): def VerifyStaticNatForPublicIp(self, ipaddressid, natrulestatus): """ List public IP and verify that NAT rule status for the IP is as desired """ - publiciplist = PublicIPAddress.list(self.apiclient, id=ipaddressid, listall=True) - self.assertEqual(validateList(publiciplist)[0], PASS, "Public IP list validation failed") - self.assertEqual(publiciplist[0].isstaticnat, natrulestatus, "isstaticnat should be %s, it is %s" % - (natrulestatus, publiciplist[0].isstaticnat)) + publiciplist = PublicIPAddress.list( + self.apiclient, + id=ipaddressid, + listall=True) + self.assertEqual( + validateList(publiciplist)[0], + PASS, + "Public IP list validation failed") + self.assertEqual( + publiciplist[0].isstaticnat, + natrulestatus, + "isstaticnat should be %s, it is %s" % + (natrulestatus, + publiciplist[0].isstaticnat)) return @@ -920,49 +1257,73 @@ class TestVmNetworkOperations(cloudstackTestCase): # 7. Verify that nat rule does not exist and static nat is not enabled for # secondary IP - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) # Add secondary IPs to default NIC of VM - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - ipaddress_2 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + ipaddress_2 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) # Acquire public IP addresses in the network - try: - public_ip_1 = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) - - public_ip_2 = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) - except Exception as e: - self.fail("Exception while acquiring public ip address: %s" % e) + public_ip_1 = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) + + public_ip_2 = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) # Create Firewall and natrule for 1st IP and static nat rule for 2nd IP if value != VPC_NETWORK: - FireWallRule.create(self.apiclient,ipaddressid=public_ip_1.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - - natrule = NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=public_ip_1.ipaddress.id, - networkid=network.id, vmguestip = ipaddress_1.ipaddress) - - StaticNATRule.enable(self.apiclient, public_ip_2.ipaddress.id, virtual_machine.id, - network.id, vmguestip=ipaddress_2.ipaddress) + FireWallRule.create( + self.apiclient, + ipaddressid=public_ip_1.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + + natrule = NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=public_ip_1.ipaddress.id, + networkid=network.id, + vmguestip=ipaddress_1.ipaddress) + + StaticNATRule.enable( + self.apiclient, + public_ip_2.ipaddress.id, + virtual_machine.id, + network.id, + vmguestip=ipaddress_2.ipaddress) # Delete VM virtual_machine.delete(self.apiclient, expunge=True) @@ -1001,46 +1362,70 @@ class TestVmNetworkOperations(cloudstackTestCase): # 6. Destroy the virtual machine and recover it # 7. Verify that nat and static nat rules exist - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) - - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - ipaddress_2 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) - - try: - public_ip_1 = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) - - public_ip_2 = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) - except Exception as e: - self.fail("Exception while acquiring public ip address: %s" % e) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) + + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + ipaddress_2 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + + public_ip_1 = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) + + public_ip_2 = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) if value != VPC_NETWORK: - FireWallRule.create(self.apiclient,ipaddressid=public_ip_1.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - - natrule = NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=public_ip_1.ipaddress.id, - networkid=network.id, vmguestip = ipaddress_1.ipaddress) - - StaticNATRule.enable(self.apiclient, public_ip_2.ipaddress.id, virtual_machine.id, - network.id, vmguestip=ipaddress_2.ipaddress) + FireWallRule.create( + self.apiclient, + ipaddressid=public_ip_1.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + + natrule = NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=public_ip_1.ipaddress.id, + networkid=network.id, + vmguestip=ipaddress_1.ipaddress) + + StaticNATRule.enable( + self.apiclient, + public_ip_2.ipaddress.id, + virtual_machine.id, + network.id, + vmguestip=ipaddress_2.ipaddress) virtual_machine.delete(self.apiclient, expunge=False) virtual_machine.recover(self.apiclient) @@ -1048,7 +1433,10 @@ class TestVmNetworkOperations(cloudstackTestCase): retriesCount = 10 while True: vms = VirtualMachine.list(self.apiclient, id=virtual_machine.id) - self.assertEqual(validateList(vms)[0], PASS, "vms list validation failed") + self.assertEqual( + validateList(vms)[0], + PASS, + "vms list validation failed") if str(vms[0].state).lower() == "stopped": break elif retriesCount == 0: @@ -1057,7 +1445,10 @@ class TestVmNetworkOperations(cloudstackTestCase): retriesCount -= 1 natrules = NATRule.list(self.apiclient, id=natrule.id, listall=True) - self.assertEqual(validateList(natrules)[0], PASS, "nat rules validation failed") + self.assertEqual( + validateList(natrules)[0], + PASS, + "nat rules validation failed") self.VerifyStaticNatForPublicIp(public_ip_2.ipaddress.id, True) @@ -1078,51 +1469,78 @@ class TestVmNetworkOperations(cloudstackTestCase): # 6. Restart the network with cleanup option True # 7. Verify that nat and static nat rules exist after network restart - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) - - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - ipaddress_2 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) - - try: - public_ip_1 = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) - - public_ip_2 = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) - except Exception as e: - self.fail("Exception while acquiring public ip address: %s" % e) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) + + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + ipaddress_2 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + + public_ip_1 = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) + + public_ip_2 = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) if value != VPC_NETWORK: - FireWallRule.create(self.apiclient,ipaddressid=public_ip_1.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - - natrule = NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=public_ip_1.ipaddress.id, - networkid=network.id, vmguestip = ipaddress_1.ipaddress) - - StaticNATRule.enable(self.apiclient, public_ip_2.ipaddress.id, virtual_machine.id, - network.id, vmguestip=ipaddress_2.ipaddress) + FireWallRule.create( + self.apiclient, + ipaddressid=public_ip_1.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + + natrule = NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=public_ip_1.ipaddress.id, + networkid=network.id, + vmguestip=ipaddress_1.ipaddress) + + StaticNATRule.enable( + self.apiclient, + public_ip_2.ipaddress.id, + virtual_machine.id, + network.id, + vmguestip=ipaddress_2.ipaddress) network.restart(self.apiclient, cleanup=True) natrulelist = NATRule.list(self.apiclient, id=natrule.id, listall=True) - self.assertEqual(validateList(natrulelist)[0], PASS, "nat rules list validation failed") + self.assertEqual( + validateList(natrulelist)[0], + PASS, + "nat rules list validation failed") self.VerifyStaticNatForPublicIp(public_ip_2.ipaddress.id, True) return @@ -1142,51 +1560,78 @@ class TestVmNetworkOperations(cloudstackTestCase): # 6. Restart the network with cleanup option False # 7. Verify that nat and static nat rules exist after network restart - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) - - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - ipaddress_2 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) - - try: - public_ip_1 = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) - - public_ip_2 = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) - except Exception as e: - self.fail("Exception while acquiring public ip address: %s" % e) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) + + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + ipaddress_2 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + + public_ip_1 = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) + + public_ip_2 = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) if value != VPC_NETWORK: - FireWallRule.create(self.apiclient,ipaddressid=public_ip_1.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - - natrule = NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=public_ip_1.ipaddress.id, - networkid=network.id, vmguestip = ipaddress_1.ipaddress) - - StaticNATRule.enable(self.apiclient, public_ip_2.ipaddress.id, virtual_machine.id, - network.id, vmguestip=ipaddress_2.ipaddress) + FireWallRule.create( + self.apiclient, + ipaddressid=public_ip_1.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + + natrule = NATRule.create( + self.api_client, + virtual_machine, + self.services["natrule"], + ipaddressid=public_ip_1.ipaddress.id, + networkid=network.id, + vmguestip=ipaddress_1.ipaddress) + + StaticNATRule.enable( + self.apiclient, + public_ip_2.ipaddress.id, + virtual_machine.id, + network.id, + vmguestip=ipaddress_2.ipaddress) network.restart(self.apiclient, cleanup=False) natrulelist = NATRule.list(self.apiclient, id=natrule.id, listall=True) - self.assertEqual(validateList(natrulelist)[0], PASS, "nat rules list validation failed") + self.assertEqual( + validateList(natrulelist)[0], + PASS, + "nat rules list validation failed") self.VerifyStaticNatForPublicIp(public_ip_2.ipaddress.id, True) return @@ -1206,54 +1651,87 @@ class TestVmNetworkOperations(cloudstackTestCase): # 6. Reboot router VM # 7. Verify that nat and static nat rules exist after router restart - self.account = Account.create(self.apiclient,self.services["account"],domainid=self.domain.id) + self.account = Account.create( + self.apiclient, + self.services["account"], + domainid=self.domain.id) self.cleanup.append(self.account) network = createNetwork(self, value) - 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.account.domainid) - except Exception as e: - self.fail("vm creation failed: %s" % e) - - try: - ipaddress_1 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - ipaddress_2 = NIC.addIp(self.apiclient, id=virtual_machine.nic[0].id) - except Exception as e: - self.fail("Failed while adding secondary IP to NIC of vm %s" % virtual_machine.id) - - try: - public_ip_1 = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) - - public_ip_2 = PublicIPAddress.create(self.api_client,accountid=self.account.name, - zoneid=self.zone.id,domainid=self.account.domainid, - networkid=network.id, vpcid = network.vpcid if value == VPC_NETWORK else None) - except Exception as e: - self.fail("Exception while acquiring public ip address: %s" % e) + virtual_machine = VirtualMachine.create( + self.apiclient, + self.services["virtual_machine"], + networkids=[ + network.id], + serviceofferingid=self.service_offering.id, + accountid=self.account.name, + domainid=self.account.domainid) + + ipaddress_1 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + ipaddress_2 = NIC.addIp( + self.apiclient, + id=virtual_machine.nic[0].id) + + public_ip_1 = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) + + public_ip_2 = PublicIPAddress.create( + self.api_client, + accountid=self.account.name, + zoneid=self.zone.id, + domainid=self.account.domainid, + networkid=network.id, + vpcid=network.vpcid if value == VPC_NETWORK else None) if value != VPC_NETWORK: - FireWallRule.create(self.apiclient,ipaddressid=public_ip_1.ipaddress.id, - protocol='TCP', cidrlist=[self.services["fwrule"]["cidr"]], - startport=self.services["fwrule"]["startport"],endport=self.services["fwrule"]["endport"]) - - natrule = NATRule.create(self.api_client, virtual_machine, - self.services["natrule"],ipaddressid=public_ip_1.ipaddress.id, - networkid=network.id, vmguestip = ipaddress_1.ipaddress) - - StaticNATRule.enable(self.apiclient, public_ip_2.ipaddress.id, virtual_machine.id, - network.id, vmguestip=ipaddress_2.ipaddress) - - routers = Router.list(self.apiclient, networkid=network.id, listall=True) - self.assertEqual(validateList(routers)[0], PASS, "routers list validation failed") + FireWallRule.create( + self.apiclient, + ipaddressid=public_ip_1.ipaddress.id, + protocol='TCP', + cidrlist=[ + self.services["fwrule"]["cidr"]], + startport=self.services["fwrule"]["startport"], + endport=self.services["fwrule"]["endport"]) + + natrule = NATRule.create( +