# Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. """ Tests for Portable public IP Ranges feature Test Plan: https://cwiki.apache.org/confluence/display/CLOUDSTACK/Portable+IP+Test+Execution Feature Specifications: https://cwiki.apache.org/confluence/display/CLOUDSTACK/portable+public+IP """ from marvin.cloudstackTestCase import cloudstackTestCase from marvin.lib.utils import cleanup_resources from marvin.lib.base import (VirtualMachine, PublicIPAddress, Network, NetworkOffering, ServiceOffering, NATRule, Account, PortablePublicIpRange, StaticNATRule, FireWallRule) from marvin.lib.common import (get_zone, get_template, get_domain, get_region, get_pod, isIpInDesiredState) from netaddr import IPAddress from marvin.sshClient import SshClient from nose.plugins.attrib import attr class TestCreatePortablePublicIpRanges(cloudstackTestCase): """Test Create Portable IP Ranges """ @classmethod def setUpClass(cls): cls.testClient = super(TestCreatePortablePublicIpRanges, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() # Fill testdata from the external config file cls.testdata = cls.testClient.getParsedTestDataConfig() # Get Zone, Domain and templates cls.region = get_region(cls.api_client) cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.pod = get_pod(cls.api_client, cls.zone.id) cls.testdata['mode'] = cls.zone.networktype cls.testdata["regionid"] = cls.region.id cls._cleanup = [] return @classmethod def tearDownClass(cls): try: #Cleanup resources used cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.cleanup = [] return def tearDown(self): try: #Clean up, terminate the resources created cleanup_resources(self.apiclient, self.cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return @attr(tags=["advanced"], required_hardware="false") def test_create_portable_ip_range(self): """Test create new portable ip range """ # 1. Create new portable ip range with root admin api # 2. Portable ip range should be created successfully self.testdata["configurableData"]["portableIpRange"]["regionid"] = self.region.id try: #create new portable ip range new_portable_ip_range = PortablePublicIpRange.create(self.apiclient, self.testdata["configurableData"]["portableIpRange"]) self.cleanup.append(new_portable_ip_range) except Exception as e: self.fail("Failed to create portable IP range: %s" % e) return @attr(tags=["advanced"], required_hardware="false") def test_create_portable_ip_range_non_root_admin(self): """Test create new portable ip range with non admin root account """ # 1. Create new portable ip range with non root admin api client # 2. Portable ip range should not be created try: self.account = Account.create( self.apiclient, self.testdata["account"], domainid=self.domain.id ) self.cleanup.append(self.account) self.api_client_user = self.testClient.getUserApiClient( UserName=self.account.name, DomainName=self.account.domain ) self.testdata["configurableData"]["portableIpRange"]["regionid"] = self.region.id self.debug("Trying to create portable ip range with non root-admin api client, should raise exception") with self.assertRaises(Exception): portable_ip_range = PortablePublicIpRange.create(self.api_client_user, self.testdata["configurableData"]["portableIpRange"]) self.cleanup.append(portable_ip_range) except Exception as e: self.fail(e) return @attr(tags=["advanced"], required_hardware="false") def test_create_portable_ip_range_invalid_region(self): """Test create portable ip range with invalid region id""" # 1. Try to create new portable ip range with invalid region id # 2. Portable ip range creation should fail self.testdata["configurableData"]["portableIpRange"]["regionid"] = -1 #create new portable ip range self.debug("Trying to create portable ip range with wrong region id") with self.assertRaises(Exception): portable_ip_range = PortablePublicIpRange.create(self.apiclient, self.testdata["configurableData"]["portableIpRange"]) self.cleanup.append(portable_ip_range) return class TestDeletePortablePublicIpRanges(cloudstackTestCase): """Test delete Portable IP Ranges """ @classmethod def setUpClass(cls): cls.testClient = super(TestDeletePortablePublicIpRanges, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() # Fill testdata from the external config file cls.testdata = cls.testClient.getParsedTestDataConfig() # Get Zone, Domain and templates cls.region = get_region(cls.api_client) cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.pod = get_pod(cls.api_client, cls.zone.id) cls.testdata['mode'] = cls.zone.networktype cls.testdata["regionid"] = cls.region.id cls._cleanup = [] return @classmethod def tearDownClass(cls): try: #Cleanup resources used cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.testdata["configurableData"]["portableIpRange"]["regionid"] = self.region.id #create new portable ip range self.portable_ip_range = PortablePublicIpRange.create(self.apiclient, self.testdata["configurableData"]["portableIpRange"]) self.cleanup = [] return def tearDown(self): try: #Clean up, terminate the resources created cleanup_resources(self.apiclient, self.cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return @attr(tags=["advanced"], required_hardware="false") def test_delete_portable_ip_range(self): """Test delete ip range """ # 1. Try to delete the created range with root admin api client # 2. Portable range should be deleted successfully self.portable_ip_range.delete(self.apiclient) return @attr(tags=["advanced"], required_hardware="false") def test_delete_portable_ip_range_non_root_admin(self): """Test delete ip range - non admin root """ # 1. Try to delete the created range with non root admin api client # 2. Portable range deletion should fail try: self.account = Account.create( self.apiclient, self.testdata["account"], domainid=self.domain.id ) self.cleanup.append(self.account) self.api_client_user = self.testClient.getUserApiClient( UserName=self.account.name, DomainName=self.account.domain ) except Exception as e: self.fail(e) try: with self.assertRaises(Exception): self.portable_ip_range.delete(self.api_client_user) except Exception as e: self.fail(e) finally: self.portable_ip_range.delete(self.apiclient) return @attr(tags=["advanced"], required_hardware="false") def test_delete_portable_ip_range_in_use(self): """Test delete ip range """ # 1. Associate a portable ip # 2. Try to delete the portable ip range with root admin api client # 3. Portable ip range should not be deleted unless currently used ip is disassociated try: self.account = Account.create( self.apiclient, self.testdata["account"], domainid=self.domain.id ) self.cleanup.append(self.account) self.network_offering = NetworkOffering.create( self.apiclient, self.testdata["isolated_network_offering"], conservemode=False ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.network = Network.create( self.apiclient, self.testdata["network"], accountid=self.account.name, domainid=self.account.domainid, networkofferingid=self.network_offering.id, zoneid=self.zone.id ) portableip = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id, isportable=True ) except Exception as e: self.fail(e) try: with self.assertRaises(Exception): self.debug("Trying to Delete portable ip range with root-admin api, this should fail") self.portable_ip_range.delete(self.apiclient) except Exception as e: self.fail(e) finally: self.debug("Disassociating portable ip") portableip.delete(self.apiclient) self.debug("Deleting portable ip range") self.portable_ip_range.delete(self.apiclient) return class TestListPortablePublicIpRanges(cloudstackTestCase): """Test List Portable IP Ranges """ @classmethod def setUpClass(cls): cls.testClient = super(TestListPortablePublicIpRanges, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() # Fill testdata from the external config file cls.testdata = cls.testClient.getParsedTestDataConfig() # Get Zone, Domain and templates cls.region = get_region(cls.api_client) cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.pod = get_pod(cls.api_client, cls.zone.id) cls.testdata['mode'] = cls.zone.networktype cls.testdata["regionid"] = cls.region.id cls._cleanup = [] return @classmethod def tearDownClass(cls): try: #Cleanup resources used cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.testdata["configurableData"]["portableIpRange"]["regionid"] = self.region.id self.debug("Creating new portable IP range with startip:%s and endip:%s" % (str(self.testdata["configurableData"]["portableIpRange"]["startip"]), str(self.testdata["configurableData"]["portableIpRange"]["endip"]))) #create new portable ip range self.portable_ip_range = PortablePublicIpRange.create(self.apiclient, self.testdata["configurableData"]["portableIpRange"]) self.debug("Created new portable IP range with startip:%s and endip:%s and id:%s" % (self.portable_ip_range.startip, self.portable_ip_range.endip, self.portable_ip_range.id)) self.cleanup = [self.portable_ip_range, ] return def tearDown(self): try: #Clean up, terminate the resources created cleanup_resources(self.apiclient, self.cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return @attr(tags=["advanced"], required_hardware="false") def test_list_portable_ip_range(self): """Test list portable ip ranges """ # 1. Create new portable ip range # 2. Try to list ip ranges with root admin api client # 3. Portable ip ranges should list properly list_portable_ip_range = PortablePublicIpRange.list(self.apiclient, id=self.portable_ip_range.id) self.assertEqual( isinstance(list_portable_ip_range, list), True, "List portable IP ranges should not return an empty response" ) portable_ip_range = list_portable_ip_range[0] self.assertEqual(str(portable_ip_range.startip), str(self.testdata["configurableData"]["portableIpRange"]["startip"]), "Listed startip not matching with the startip of created public ip range") self.assertEqual(str(portable_ip_range.endip), str(self.testdata["configurableData"]["portableIpRange"]["endip"]), "Listed endip not matching with the endip of created public ip range") self.assertEqual(str(portable_ip_range.gateway), str(self.testdata["configurableData"]["portableIpRange"]["gateway"]), "Listed gateway not matching with the gateway of created public ip range") self.assertEqual(str(portable_ip_range.netmask), str(self.testdata["configurableData"]["portableIpRange"]["netmask"]), "Listed netmask not matching with the netmask of created public ip range") return @attr(tags=["advanced"], required_hardware="false") def test_list_portable_ip_range_non_root_admin(self): """Test list portable ip ranges with non admin root account """ # 1. Create new portable ip range # 2. Try to list ip ranges with root non admin api client # 3. Portable ip ranges listing should fail self.account = Account.create( self.apiclient, self.testdata["account"], domainid=self.domain.id ) self.cleanup.append(self.account) self.api_client_user = self.testClient.getUserApiClient( UserName=self.account.name, DomainName=self.account.domain ) self.debug("Trying to list portable ip ranges with non root-admin api, should raise exception") with self.assertRaises(Exception): PortablePublicIpRange.list(self.api_client_user, id=self.portable_ip_range.id) return class TestAssociatePublicIp(cloudstackTestCase): """Test associate Portable IP/ non portable public ip """ @classmethod def setUpClass(cls): cls.testClient = super(TestAssociatePublicIp, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() # Fill testdata from the external config file cls.testdata = cls.testClient.getParsedTestDataConfig() # Get Zone, Domain and templates cls.region = get_region(cls.api_client) cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.pod = get_pod(cls.api_client, cls.zone.id) cls.testdata['mode'] = cls.zone.networktype cls.testdata["regionid"] = cls.region.id template = get_template( cls.api_client, cls.zone.id, cls.testdata["ostype"] ) # Set Zones and disk offerings cls.testdata["small"]["zoneid"] = cls.zone.id cls.testdata["small"]["template"] = template.id cls.account = Account.create( cls.api_client, cls.testdata["account"], domainid=cls.domain.id, admin=True ) cls._cleanup = [cls.account, ] cls.network_offering = NetworkOffering.create( cls.api_client, cls.testdata["isolated_network_offering"], conservemode=False ) # Enable Network offering cls.network_offering.update(cls.api_client, state='Enabled') cls.network = Network.create( cls.api_client, cls.testdata["network"], accountid=cls.account.name, domainid=cls.account.domainid, networkofferingid=cls.network_offering.id, zoneid=cls.zone.id ) return @classmethod def tearDownClass(cls): try: # Disable Network offering cls.network_offering.update(cls.api_client, state='Disabled') #Cleanup resources used cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.cleanup = [] self.testdata["configurableData"]["portableIpRange"]["regionid"] = self.region.id #create new portable ip range self.portable_ip_range = PortablePublicIpRange.create(self.apiclient, self.testdata["configurableData"]["portableIpRange"]) self.cleanup.append(self.portable_ip_range) return def tearDown(self): try: #Clean up, terminate the resources created self.network_offering.update(self.apiclient, state='Disabled') cleanup_resources(self.apiclient, self.cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return @attr(tags=["advanced"], required_hardware="false") def test_associate_ip_address(self): """ Test assocoate public ip address """ # 1. Create new portable ip range # 2. Create a network and associate public ip without mentioning (isportable) # 3. Create a network and associate public ip with isportable=False # 4. Create a network and associate public ip with isPortable=True # 5. All three public ip associations should succeed self.debug("Associating default public ip address with network: %s" % self.network.id) publicipaddress = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id ) self.debug("Associated default public ip address: %s" % publicipaddress.ipaddress.ipaddress) self.debug("Associating public ip address with network: %s with isportable=False" % self.network.id) publicipaddressnotportable = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id, isportable=False ) self.debug("Associated public ip address (not portable): %s" % publicipaddressnotportable.ipaddress.ipaddress) publicipaddressnotportable.delete(self.apiclient) self.debug("Associating public ip address with network: %s with isportable=True" % self.network.id) publicipaddressportable = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id, isportable=True ) self.debug("Associated public ip address (portable): %s" % publicipaddressportable.ipaddress.ipaddress) publicipaddressportable.delete(self.apiclient) return @attr(tags=["advanced"], required_hardware="false") def test_associate_ip_address_invalid_zone(self): """ Test Associate IP with invalid zone id """ # 1. Create new portable ip range # 2. try to associate a portable ip with invalid region id # 3. IP association should fail self.debug("Trying to associate portable public ip with invalid zone id, this should fail") with self.assertRaises(Exception): publicipaddress = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid = -1, domainid=self.account.domainid, regionid = self.region.id, isportable=True ) publicipaddress.delete(self.apiclient) return # @attr(tags=["advanced"], required_hardware="true") @attr(tags=["TODO"], required_hardware="true") def test_associate_ip_address_services_enable_disable(self): """ Test enabling and disabling NAT, Firewall services on portable ip """ # 1. Create new portable ip range # 2. Associate a portable ip # 3. Enable NAT and Firewall rules on this portable ip # 4. Disable NAT and Firewall rules created # 5. Enabling and disabling ofthe rules should be successful self.service_offering = ServiceOffering.create( self.apiclient, self.testdata["service_offering"] ) self.cleanup.append(self.service_offering) try: self.debug("DeployingVirtual Machine") self.virtual_machine = VirtualMachine.create( self.apiclient, self.testdata["small"], accountid=self.account.name, domainid=self.account.domainid, serviceofferingid=self.service_offering.id, networkids = [self.network.id], mode=self.testdata['mode'] ) self.debug("Created virtual machine instance: %s with ssh_ip: %s" % (self.virtual_machine.id, self.virtual_machine.ssh_ip)) except Exception as e: self.fail("Exception while deploying vm : %s" % e) portableip = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id, isportable=True ) self.debug("created public ip address (portable): %s" % portableip.ipaddress.ipaddress) response = isIpInDesiredState(self.apiclient, portableip.ipaddress.id, state="allocated") exceptionOccurred = response[0] ipInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccurred or (not ipInDesiredState)): portableip.delete(self.apiclient) self.fail(exceptionMessage) try: # Open up firewall port for SSH self.debug("Opening firewall on the portable public ip") fw_rule = FireWallRule.create( self.apiclient, ipaddressid=portableip.ipaddress.id, protocol=self.testdata["natrule"]["protocol"], cidrlist=["0.0.0.0/0"], startport=self.testdata["natrule"]["publicport"], endport=self.testdata["natrule"]["publicport"] ) #Create NAT rule self.debug("Creating NAT rule on the portable public ip") nat_rule = NATRule.create( self.apiclient, self.virtual_machine, self.testdata["natrule"], portableip.ipaddress.id ) except Exception as e: portableip.delete(self.apiclient) self.fail("Error: %s" % e) try: self.debug("Trying to SSH to ip: %s" % portableip.ipaddress.ipaddress) SshClient(portableip.ipaddress.ipaddress, self.testdata['natrule']["publicport"], self.virtual_machine.username, self.virtual_machine.password ) except Exception as e: self.fail("Exception while SSHing : %s" % e) finally: self.debug("Deleting firewall rule") fw_rule.delete(self.apiclient) self.debug("Deleting NAT rule") nat_rule.delete(self.apiclient) self.debug("disassocoating portable ip: %s" % portableip.ipaddress.ipaddress) portableip.delete(self.apiclient) return @attr(tags=["advanced"], required_hardware="false") def test_associate_ip_address_no_free_ip(self): """ Test assocoate public ip address """ # 1. Create new portable ip range # 2. Create a network and associate all available portbale public ips # 5. Try to associate portable ip, it should fail associatedipaddresses = [] startip_int = int(IPAddress(self.portable_ip_range.startip)) endip_int = int(IPAddress(self.portable_ip_range.endip)) totalportableips = ((endip_int - startip_int) + 1) self.debug(totalportableips) for x in range(0, totalportableips): self.debug("Associating public ip address with network: %s with isportable=True" % self.network.id) portableip = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id, isportable=True ) associatedipaddresses.append(portableip) self.debug("Associated public ip address (portable): %s" % portableip.ipaddress.ipaddress) self.debug("Trying to associate portable public ip when no free ips available, this should fail") with self.assertRaises(Exception): portableipaddress = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id, isportable=True ) portableipaddress.delete(self.apiclient) self.debug("Associating portable ip address failed") self.debug("Disassociating previously associated ip addresses") for x in range(0, totalportableips): associatedipaddresses[x].delete(self.apiclient) return class TestDisassociatePublicIp(cloudstackTestCase): """Test Disassociate Portable IP/ non portable IP """ @classmethod def setUpClass(cls): cls.testClient = super(TestDisassociatePublicIp, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() # Fill testdata from the external config file cls.testdata = cls.testClient.getParsedTestDataConfig() # Get Zone, Domain and templates cls.region = get_region(cls.api_client) cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.pod = get_pod(cls.api_client, cls.zone.id) cls.testdata['mode'] = cls.zone.networktype cls.testdata["regionid"] = cls.region.id template = get_template( cls.api_client, cls.zone.id, cls.testdata["ostype"] ) # Set Zones and disk offerings cls.testdata["small"]["zoneid"] = cls.zone.id cls.testdata["small"]["template"] = template.id cls._cleanup = [] cls.account = Account.create( cls.api_client, cls.testdata["account"], domainid=cls.domain.id, admin=True ) cls._cleanup.append(cls.account) cls.service_offering = ServiceOffering.create( cls.api_client, cls.testdata["service_offering"] ) cls._cleanup.append(cls.service_offering) cls.network_offering = NetworkOffering.create( cls.api_client, cls.testdata["isolated_network_offering"], conservemode=False ) # Enable Network offering cls.network_offering.update(cls.api_client, state='Enabled') cls._cleanup.append(cls.network_offering) cls.network = Network.create( cls.api_client, cls.testdata["network"], accountid=cls.account.name, domainid=cls.account.domainid, networkofferingid=cls.network_offering.id, zoneid=cls.zone.id ) cls.virtual_machine = VirtualMachine.create( cls.api_client, cls.testdata["small"], accountid=cls.account.name, domainid=cls.account.domainid, serviceofferingid=cls.service_offering.id, networkids = [cls.network.id], mode=cls.testdata['mode'] ) return @classmethod def tearDownClass(cls): try: # Disable Network offering cls.network_offering.update(cls.api_client, state='Disabled') #Cleanup resources used cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.cleanup = [] self.testdata["configurableData"]["portableIpRange"]["regionid"] = self.region.id #create new portable ip range new_portable_ip_range = PortablePublicIpRange.create(self.apiclient, self.testdata["configurableData"]["portableIpRange"]) self.cleanup.append(new_portable_ip_range) return def tearDown(self): try: #Clean up, terminate the resources created self.network_offering.update(self.apiclient, state='Disabled') cleanup_resources(self.apiclient, self.cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return @attr(tags=["advanced"], required_hardware="false") def test_disassociate_ip_address_no_services(self): """ Test disassociating portable ip """ # 1. Create new portable ip range # 2. Associate a portable ip # 3. Disassociate the portable ip with root admin api client # 4. Disassociating should be successful try: portableip = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id, isportable=True ) portableip.delete(self.apiclient) except Exception as e: raise Exception("Exception occurred: %s" % e) return @attr(tags=["advanced"], required_hardware="false") def test_disassociate_ip_address_services_enabled(self): """ Test disassociating portable ip """ # 1. Create new portable ip range # 2. Associate a portable ip # 3. Enable NAT and Firewall services on this portable IP # 4. Disassociate the portable ip with root admin api client # 5. Disassociating should be successful portableip = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id, isportable=True ) response = isIpInDesiredState(self.apiclient, portableip.ipaddress.id, state="allocated") exceptionOccurred = response[0] ipInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccurred or (not ipInDesiredState)): portableip.delete(self.apiclient) self.fail(exceptionMessage) try: # Open up firewall port for SSH self.debug("Opening firewall on the portable public ip") FireWallRule.create( self.apiclient, ipaddressid=portableip.ipaddress.id, protocol=self.testdata["natrule"]["protocol"], cidrlist=["0.0.0.0/0"], startport=self.testdata["natrule"]["publicport"], endport=self.testdata["natrule"]["publicport"] ) #Create NAT rule self.debug("Creating NAT rule on the portable public ip") NATRule.create( self.apiclient, self.virtual_machine, self.testdata["natrule"], portableip.ipaddress.id ) except Exception as e: portableip.delete(self.apiclient) self.fail("Error: %s" % e) try: portableip.delete(self.apiclient) except Exception as e: raise Exception("Exception while disassociating portable ip: %s" % e) return @attr(tags=["advanced"], required_hardware="false") def test_disassociate_ip_address_other_account(self): """ Test disassociating portable IP with non-owner account """ # 1. Create new portable ip range # 2. Associate a portable ip # 3. Try to Disassociate the portable ip with an account which is not owner of portable ip # 4. Disassociating should fail try: portableip = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id, isportable=True ) except Exception as e: self.fail("Failed to create portable ip: %s" % e) try: self.otherAccount = Account.create( self.apiclient, self.testdata["account"], domainid=self.domain.id ) self.cleanup.append(self.otherAccount) self.apiclientOtherAccount = self.testClient.getUserApiClient( UserName=self.otherAccount.name, DomainName=self.otherAccount.domain ) # Trying to disassociate portable ip using # api client of other account than the one # used to create portable ip with self.assertRaises(Exception): portableip.delete(self.apiclientOtherAccount) # Disassociate IP using api client of account used to create it portableip.delete(self.apiclient) except Exception as e: self.fail("Exception while disassociating portable ip: %s" % e) return class TestDeleteAccount(cloudstackTestCase): """ Test Delete Account having portable ip """ @classmethod def setUpClass(cls): cls.testClient = super(TestDeleteAccount, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() # Fill testdata from the external config file cls.testdata = cls.testClient.getParsedTestDataConfig() # Get Zone, Domain and templates cls.region = get_region(cls.api_client) cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.pod = get_pod(cls.api_client, cls.zone.id) cls.testdata['mode'] = cls.zone.networktype cls.testdata["regionid"] = cls.region.id template = get_template( cls.api_client, cls.zone.id, cls.testdata["ostype"] ) # Set Zones and disk offerings cls.testdata["small"]["zoneid"] = cls.zone.id cls.testdata["small"]["template"] = template.id cls._cleanup = [] return @classmethod def tearDownClass(cls): try: #Cleanup resources used cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.cleanup = [] try: self.account = Account.create( self.apiclient, self.testdata["account"], domainid=self.domain.id, admin=True ) self.cleanup.append(self.account) self.testdata["configurableData"]["portableIpRange"]["regionid"] = self.region.id #create new portable ip range new_portable_ip_range = PortablePublicIpRange.create(self.apiclient, self.testdata["configurableData"]["portableIpRange"]) self.cleanup.append(new_portable_ip_range) self.network_offering = NetworkOffering.create( self.apiclient, self.testdata["isolated_network_offering"], conservemode=False ) # Enable Network offering self.network_offering.update(self.apiclient, state='Enabled') self.network = Network.create( self.apiclient, self.testdata["network"], accountid=self.account.name, domainid=self.account.domainid, networkofferingid=self.network_offering.id, zoneid=self.zone.id ) self.cleanup.append(self.network_offering) except Exception as e: self.fail("Exception in setupClass: %s" % e) return def tearDown(self): try: #Clean up, terminate the resources created cleanup_resources(self.apiclient, self.cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return @attr(tags=["advanced"], required_hardware="false") def test_delete_account_services_disabled(self): """ test delete account with portable ip with no services enabled """ # 1. Associate a portable ip to an account # 2. Delete account # 3. Account should get deleted successfully portableip = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id, isportable=True ) self.account.delete(self.apiclient) list_publicips = PublicIPAddress.list(self.apiclient, id=portableip.ipaddress.id) self.assertEqual(list_publicips, None, "List of ip addresses should be empty") return @attr(tags=["advanced"], required_hardware="false") def test_delete_account_services_enabled(self): """ test delete account with portable ip with PF and firewall services enabled """ # 1. Associate a portable ip to an account # 2. Enabled PF and Firewall rules on this IP # 3. Delete account # 4. Account should get deleted successfully self.service_offering = ServiceOffering.create( self.apiclient, self.testdata["service_offering"] ) self.cleanup.append(self.service_offering) self.debug("Deploying Virtual Machine") self.virtual_machine = VirtualMachine.create( self.apiclient, self.testdata["small"], accountid=self.account.name, domainid=self.account.domainid, serviceofferingid=self.service_offering.id, mode=self.testdata['mode'] ) self.debug("Created virtual machine instance: %s" % self.virtual_machine.id) portableip = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network.id, isportable=True ) self.debug("created public ip address (portable): %s" % portableip.ipaddress.ipaddress) response = isIpInDesiredState(self.apiclient, portableip.ipaddress.id, state="allocated") exceptionOccurred = response[0] ipInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccurred or (not ipInDesiredState)): portableip.delete(self.apiclient) self.account.delete(self.apiclient) self.fail(exceptionMessage) try: # Open up firewall port for SSH self.debug("Opening firewall on the portable public ip") FireWallRule.create( self.apiclient, ipaddressid=portableip.ipaddress.id, protocol=self.testdata["natrule"]["protocol"], cidrlist=["0.0.0.0/0"], startport=self.testdata["natrule"]["publicport"], endport=self.testdata["natrule"]["publicport"] ) #Create NAT rule self.debug("Creating NAT rule on the portable public ip") NATRule.create( self.apiclient, self.virtual_machine, self.testdata["natrule"], portableip.ipaddress.id ) except Exception as e: portableip.delete(self.apiclient) self.account.delete(self.apiclient) self.fail("Error %s" % e) self.debug("Deleting account: %s :" % self.account.name) self.account.delete(self.apiclient) self.debug("Trying to list the ip address associated with deleted account, \ should throw exception") list_publicips = PublicIPAddress.list(self.apiclient, id=portableip.ipaddress.id) self.assertEqual(list_publicips, None, "List of ip addresses should be empty") return class TestPortableIpTransferAcrossNetworks(cloudstackTestCase): """Test Transfer Portable IP Across Networks """ @classmethod def setUpClass(cls): cls.testClient = super(TestPortableIpTransferAcrossNetworks, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() # Fill testdata from the external config file cls.testdata = cls.testClient.getParsedTestDataConfig() # Get Zone, Domain and templates cls.region = get_region(cls.api_client) cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.pod = get_pod(cls.api_client, cls.zone.id) cls.testdata['mode'] = cls.zone.networktype cls.testdata["regionid"] = cls.region.id template = get_template( cls.api_client, cls.zone.id, cls.testdata["ostype"] ) # Set Zones and disk offerings cls.testdata["small"]["zoneid"] = cls.zone.id cls.testdata["small"]["template"] = template.id cls._cleanup = [] # Set Zones and Network offerings cls.account = Account.create( cls.api_client, cls.testdata["account"], domainid=cls.domain.id, admin=True ) cls._cleanup.append(cls.account) cls.network_offering = NetworkOffering.create( cls.api_client, cls.testdata["isolated_network_offering"], conservemode=False ) cls._cleanup.append(cls.network_offering) # Enable Network offering cls.network_offering.update(cls.api_client, state='Enabled') cls.service_offering = ServiceOffering.create( cls.api_client, cls.testdata["service_offering"] ) cls.network1 = Network.create( cls.api_client, cls.testdata["network"], accountid=cls.account.name, domainid=cls.account.domainid, networkofferingid=cls.network_offering.id, zoneid=cls.zone.id ) cls.virtual_machine1 = VirtualMachine.create( cls.api_client, cls.testdata["small"], accountid=cls.account.name, domainid=cls.account.domainid, serviceofferingid=cls.service_offering.id, networkids = [cls.network1.id], ) cls.network2 = Network.create( cls.api_client, cls.testdata["network"], accountid=cls.account.name, domainid=cls.account.domainid, networkofferingid=cls.network_offering.id, zoneid=cls.zone.id ) cls.virtual_machine2 = VirtualMachine.create( cls.api_client, cls.testdata["small"], accountid=cls.account.name, domainid=cls.account.domainid, serviceofferingid=cls.service_offering.id, networkids = [cls.network2.id], ) return @classmethod def tearDownClass(cls): try: #Cleanup resources used cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.testdata["configurableData"]["portableIpRange"]["regionid"] = self.region.id #create new portable ip range self.portable_ip_range = PortablePublicIpRange.create(self.apiclient, self.testdata["configurableData"]["portableIpRange"]) self.cleanup = [self.portable_ip_range, ] return def tearDown(self): try: #Clean up, terminate the resources created cleanup_resources(self.apiclient, self.cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return @attr(tags=["advanced"], required_hardware="true") def test_list_portable_ip_range_non_root_admin(self): """Test list portable ip ranges with non admin root account """ # 1. Create new network 1 and associate portable IP 1 # 2. Have at least 1 VM in network1 # 3. Create a new network 2 and at least 1 VM in network 2 # 2. enable static NAT on portable IP 1 with a VM in network 2 # 3. SSH to the VM in network 2 portableip = PublicIPAddress.create( self.apiclient, accountid=self.account.name, zoneid=self.zone.id, domainid=self.account.domainid, networkid=self.network1.id, isportable=True ) response = isIpInDesiredState(self.apiclient, portableip.ipaddress.id, state="allocated") exceptionOccurred = response[0] ipInDesiredState = response[1] exceptionMessage = response[2] if (exceptionOccurred or (not ipInDesiredState)): portableip.delete(self.apiclient) self.fail(exceptionMessage) self.debug("created public ip address (portable): %s" % portableip.ipaddress.ipaddress) #Create NAT rule self.debug("Creating NAT rule on the portable public ip") try: # Enable Static NAT for VM StaticNATRule.enable( self.apiclient, portableip.ipaddress.id, self.virtual_machine2.id, networkid=self.network2.id ) # Open up firewall port for SSH self.debug("Opening firewall on the portable public ip") FireWallRule.create( self.apiclient, ipaddressid=portableip.ipaddress.id, protocol=self.testdata["natrule"]["protocol"], cidrlist=["0.0.0.0/0"], startport=self.testdata["natrule"]["publicport"], endport=self.testdata["natrule"]["publicport"] ) except Exception as e: portableip.delete(self.apiclient) self.fail("Error: %s" % e) static_nat_list = PublicIPAddress.list( self.apiclient, associatednetworkid=self.network2.id, listall=True, isstaticnat=True, ipaddress=portableip.ipaddress.ipaddress, ) self.assertEqual( isinstance(static_nat_list, list), True, "List Public IP should return a valid static NAT info that was created on portable ip" ) self.assertTrue( static_nat_list[0].ipaddress == portableip.ipaddress.ipaddress and static_nat_list[0].virtualmachineid==self.virtual_machine2.id, "There is some issue in transferring portable ip {} across networks".format(portableip.ipaddress.ipaddress) ) try: self.debug("Trying to SSH to ip: %s" % portableip.ipaddress.ipaddress) SshClient(portableip.ipaddress.ipaddress, self.testdata['natrule']["publicport"], self.virtual_machine2.username, self.virtual_machine2.password ) except Exception as e: self.fail("Probably hit issue: CLOUDSTACK-10078, " "Exception while SSHing : %s" % e) finally: self.debug("disassociating portable ip: %s" % portableip.ipaddress.ipaddress) portableip.delete(self.apiclient)