| # 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 of user-private gateway | |
| """ | |
| import logging | |
| import time | |
| from nose.plugins.attrib import attr | |
| from marvin.cloudstackTestCase import cloudstackTestCase | |
| from marvin.lib.utils import cleanup_resources, random_gen | |
| from marvin.lib.base import (Account, | |
| Domain, | |
| Project, | |
| Configurations, | |
| ServiceOffering, | |
| Zone, | |
| Network, | |
| NetworkOffering, | |
| VPC, | |
| VpcOffering, | |
| PrivateGateway) | |
| from marvin.lib.common import (get_domain, | |
| get_zone, | |
| get_template) | |
| NETWORK_STATE_ALLOCATED = "Allocated" | |
| NETWORK_STATE_IMPLEMENTED = "Implemented" | |
| NETWORK_STATE_SETUP = "Setup" | |
| NETWORK_STATE_REMOVED = "Removed" | |
| class TestUserPrivateGateways(cloudstackTestCase): | |
| """ | |
| Test user-shared networks | |
| """ | |
| @classmethod | |
| def setUpClass(cls): | |
| cls.testClient = super( | |
| TestUserPrivateGateways, | |
| cls).getClsTestClient() | |
| cls.apiclient = cls.testClient.getApiClient() | |
| cls.apiclient = cls.testClient.getApiClient() | |
| cls.services = cls.testClient.getParsedTestDataConfig() | |
| zone = get_zone(cls.apiclient, cls.testClient.getZoneForTests()) | |
| cls.zone = Zone(zone.__dict__) | |
| cls.template = get_template(cls.apiclient, cls.zone.id) | |
| cls._cleanup = [] | |
| cls.logger = logging.getLogger("TestUserPrivateGateways") | |
| cls.stream_handler = logging.StreamHandler() | |
| cls.logger.setLevel(logging.DEBUG) | |
| cls.logger.addHandler(cls.stream_handler) | |
| cls.domain = get_domain(cls.apiclient) | |
| # Create small service offering | |
| cls.service_offering = ServiceOffering.create( | |
| cls.apiclient, | |
| cls.services["service_offerings"]["small"] | |
| ) | |
| cls._cleanup.append(cls.service_offering) | |
| # Create network offering for isolated networks | |
| cls.network_offering_isolated = NetworkOffering.create( | |
| cls.apiclient, | |
| cls.services["network_offering"] | |
| ) | |
| cls.network_offering_isolated.update(cls.apiclient, state='Enabled') | |
| cls._cleanup.append(cls.network_offering_isolated) | |
| # Create vpc offering | |
| cls.vpc_offering = VpcOffering.create( | |
| cls.apiclient, | |
| cls.services["vpc_offering_multi_lb"]) | |
| cls.vpc_offering.update(cls.apiclient, state='Enabled') | |
| cls._cleanup.append(cls.vpc_offering) | |
| # Create network offering for vpc tiers | |
| cls.network_offering_vpc = NetworkOffering.create( | |
| cls.apiclient, | |
| cls.services["nw_offering_isolated_vpc"], | |
| conservemode=False | |
| ) | |
| cls.network_offering_vpc.update(cls.apiclient, state='Enabled') | |
| cls._cleanup.append(cls.network_offering_vpc) | |
| # Create sub-domain | |
| cls.sub_domain = Domain.create( | |
| cls.apiclient, | |
| cls.services["acl"]["domain1"] | |
| ) | |
| cls._cleanup.append(cls.sub_domain) | |
| # Create domain admin and normal user | |
| cls.domain_admin = Account.create( | |
| cls.apiclient, | |
| cls.services["acl"]["accountD1A"], | |
| admin=True, | |
| domainid=cls.sub_domain.id | |
| ) | |
| cls._cleanup.append(cls.domain_admin) | |
| cls.normal_user = Account.create( | |
| cls.apiclient, | |
| cls.services["acl"]["accountD1B"], | |
| domainid=cls.sub_domain.id | |
| ) | |
| cls._cleanup.append(cls.normal_user) | |
| # Create project | |
| cls.project = Project.create( | |
| cls.apiclient, | |
| cls.services["project"], | |
| account=cls.domain_admin.name, | |
| domainid=cls.domain_admin.domainid | |
| ) | |
| cls._cleanup.append(cls.project) | |
| # Create api clients for domain admin and normal user | |
| cls.domainadmin_user = cls.domain_admin.user[0] | |
| cls.domainapiclient = cls.testClient.getUserApiClient( | |
| cls.domainadmin_user.username, cls.sub_domain.name | |
| ) | |
| cls.normaluser_user = cls.normal_user.user[0] | |
| cls.normaluser_apiclient = cls.testClient.getUserApiClient( | |
| cls.normaluser_user.username, cls.sub_domain.name | |
| ) | |
| @classmethod | |
| def tearDownClass(cls): | |
| super(TestUserPrivateGateways, cls).tearDownClass() | |
| def setUp(self): | |
| self.cleanup = [] | |
| def tearDown(self): | |
| super(TestUserPrivateGateways, self).tearDown() | |
| def delete_network(self, network, apiclient, expected=True): | |
| result = True | |
| try: | |
| Network.delete( | |
| network, | |
| apiclient, | |
| ) | |
| except Exception as ex: | |
| result = False | |
| if expected: | |
| self.fail(f"Failed to remove Shared network, but expected to succeed : {ex}") | |
| if result and not expected: | |
| self.fail("network is removed successfully, but expected to fail") | |
| def create_isolated_network_for_account(self, apiclient, domain, account, project, expected=True): | |
| self.services["network"]["name"] = "Test Network Isolated - " + random_gen() | |
| domain_id = None | |
| account_name = None | |
| project_id = None | |
| if domain: | |
| domain_id = domain.id | |
| if account: | |
| account_name = account.name | |
| if project: | |
| project_id = project.id | |
| try: | |
| network = Network.create( | |
| apiclient, | |
| self.services["network"], | |
| domainid=domain_id, | |
| accountid=account_name, | |
| projectid=project_id, | |
| networkofferingid=self.network_offering_isolated.id, | |
| zoneid=self.zone.id | |
| ) | |
| except Exception as ex: | |
| network = None | |
| if expected: | |
| self.fail(f"Failed to create Isolated network, but expected to succeed : {ex}") | |
| if network and not expected: | |
| self.fail("Isolated network is created successfully, but expected to fail") | |
| return network | |
| def check_network_state(self, apiclient, network, project, expected_state): | |
| project_id = None | |
| if project: | |
| project_id = project.id | |
| networks = Network.list( | |
| apiclient, | |
| listall=True, | |
| projectid=project_id, | |
| id=network.id | |
| ) | |
| if isinstance(networks, list) and len(networks) > 0: | |
| if expected_state == NETWORK_STATE_REMOVED: | |
| self.fail("Found the network, but expected to fail") | |
| if networks[0].state != expected_state: | |
| self.fail(f"Expect network state is {expected_state}, but actual state is {networks[0].state}") | |
| elif expected_state != NETWORK_STATE_REMOVED: | |
| self.fail("Failed to find the network, but expected to succeed") | |
| def create_vpc_for_account(self, apiclient, domain, account, project): | |
| self.services["vpc"]["name"] = "Test VPC - " + random_gen() | |
| self.services["vpc"]["cidr"] = "10.1.0.0/20" | |
| domain_id = None | |
| account_name = None | |
| if domain: | |
| domain_id = domain.id | |
| if account: | |
| account_name = account.name | |
| project_id = None | |
| if project: | |
| project_id = project.id | |
| vpc = VPC.create( | |
| apiclient, | |
| self.services["vpc"], | |
| domainid=domain_id, | |
| accountid=account_name, | |
| projectid=project_id, | |
| vpcofferingid=self.vpc_offering.id, | |
| zoneid=self.zone.id, | |
| start=False | |
| ) | |
| return vpc | |
| def create_vpc_tier_for_account(self, apiclient, vpc, project=None, gateway = '10.1.1.1'): | |
| self.services["network"]["name"] = "Test VPC tier - " + random_gen() | |
| project_id = None | |
| if project: | |
| project_id = project.id | |
| vpc_tier = Network.create( | |
| apiclient, | |
| self.services["network"], | |
| networkofferingid=self.network_offering_vpc.id, | |
| zoneid=self.zone.id, | |
| projectid=project_id, | |
| gateway=gateway, | |
| netmask="255.255.255.0", | |
| vpcid=vpc.id | |
| ) | |
| return vpc_tier | |
| def create_vpc_private_gateway(self, apiclient, vpc, vlan_id, associated_network=None, expected=True): | |
| self.services["private_gateway"]["name"] = "Test Network Isolated - " + random_gen() | |
| associated_network_id = None | |
| if associated_network: | |
| associated_network_id = associated_network.id | |
| private_gateway = None | |
| try: | |
| private_gateway = PrivateGateway.create( | |
| apiclient, | |
| vpcid=vpc.id, | |
| gateway = self.services["private_gateway"]["gateway"], | |
| ipaddress = self.services["private_gateway"]["ipaddress"], | |
| netmask = self.services["private_gateway"]["netmask"], | |
| vlan = vlan_id, | |
| associatednetworkid = associated_network_id | |
| ) | |
| except Exception as ex: | |
| private_gateway = None | |
| if expected: | |
| self.fail(f"Failed to create private gateway, but expected to succeed : {ex}") | |
| if private_gateway and not expected: | |
| self.fail("private gateway is created successfully, but expected to fail") | |
| return private_gateway | |
| def delete_vpc_private_gateway(self, apiclient, private_gateway, expected=True): | |
| result = True | |
| try: | |
| private_gateway.delete(apiclient) | |
| except Exception as ex: | |
| result = False | |
| if expected: | |
| self.fail(f"Failed to remove private gateway, but expected to succeed : {ex}") | |
| if result and not expected: | |
| self.fail("private gateway is removed successfully, but expected to fail") | |
| def delete_vpc(self, apiclient, vpc, expected=True): | |
| result = True | |
| try: | |
| vpc.delete(apiclient) | |
| except Exception as ex: | |
| result = False | |
| if expected: | |
| self.fail(f"Failed to remove VPC, but expected to succeed : {ex}") | |
| if result and not expected: | |
| self.fail("VPC is removed successfully, but expected to fail") | |
| @attr(tags=["advanced"], required_hardware="false") | |
| def test_01_create_private_gateway_with_vlan(self): | |
| """ Create private gateway with vlan """ | |
| # Create VPC | |
| vpc1 = self.create_vpc_for_account(self.apiclient, None, None, None) | |
| vpc2 = self.create_vpc_for_account(self.domainapiclient, None, None, None) | |
| vpc3 = self.create_vpc_for_account(self.normaluser_apiclient, None, None, None) | |
| vpc4 = self.create_vpc_for_account(self.domainapiclient, None, None, self.project) | |
| # Create VPC private gateway with vlan (can only be done by ROOT admin) | |
| private_gateway1 = self.create_vpc_private_gateway(self.apiclient, vpc1, 10, None, True) | |
| self.create_vpc_private_gateway(self.domainapiclient, vpc2, 11, None, False) | |
| self.create_vpc_private_gateway(self.normaluser_apiclient, vpc3, 12, None, False) | |
| self.create_vpc_private_gateway(self.domainapiclient, vpc4, 13, None, False) | |
| private_gateway2 = self.create_vpc_private_gateway(self.apiclient, vpc2, 11, None, True) | |
| private_gateway3 = self.create_vpc_private_gateway(self.apiclient, vpc3, 12, None, True) | |
| private_gateway4 = self.create_vpc_private_gateway(self.apiclient, vpc4, 13, None, True) | |
| # Delete VPC private gateway (should succeed by ROOT admin) | |
| self.delete_vpc_private_gateway(self.apiclient, private_gateway1, True) | |
| self.delete_vpc_private_gateway(self.domainapiclient, private_gateway2, False) | |
| self.delete_vpc_private_gateway(self.normaluser_apiclient, private_gateway2, False) | |
| self.delete_vpc_private_gateway(self.domainapiclient, private_gateway4, False) | |
| self.delete_vpc_private_gateway(self.apiclient, private_gateway2, True) | |
| self.delete_vpc_private_gateway(self.apiclient, private_gateway3, True) | |
| self.delete_vpc_private_gateway(self.apiclient, private_gateway4, True) | |
| # Delete VPC | |
| self.delete_vpc(self.apiclient, vpc1) | |
| self.delete_vpc(self.domainapiclient, vpc2) | |
| self.delete_vpc(self.normaluser_apiclient, vpc3) | |
| self.delete_vpc(self.domainapiclient, vpc4) | |
| @attr(tags=["advanced"], required_hardware="false") | |
| def test_02_create_private_gateway_with_associated_network(self): | |
| """ Create private gateway with associated network """ | |
| self.services["network"]["networkoffering"] = self.network_offering_isolated.id | |
| self.services["network"]["vlan"] = None | |
| # Create isolated networks | |
| isolated_network1 = self.create_isolated_network_for_account(self.apiclient, None, None, None) | |
| isolated_network2 = self.create_isolated_network_for_account(self.apiclient, self.sub_domain, self.domain_admin, None) | |
| isolated_network3 = self.create_isolated_network_for_account(self.apiclient, self.sub_domain, self.normal_user, None) | |
| isolated_network4 = self.create_isolated_network_for_account(self.apiclient, self.sub_domain, None, self.project) | |
| # Check state of isolated networks (should be Allocated) | |
| self.check_network_state(self.apiclient, isolated_network1, None, NETWORK_STATE_ALLOCATED) | |
| self.check_network_state(self.domainapiclient, isolated_network2, None, NETWORK_STATE_ALLOCATED) | |
| self.check_network_state(self.normaluser_apiclient, isolated_network3, None, NETWORK_STATE_ALLOCATED) | |
| self.check_network_state(self.domainapiclient, isolated_network4, self.project, NETWORK_STATE_ALLOCATED) | |
| # Create VPC | |
| vpc1 = self.create_vpc_for_account(self.apiclient, None, None, None) | |
| vpc2 = self.create_vpc_for_account(self.domainapiclient, None, None, None) | |
| vpc3 = self.create_vpc_for_account(self.normaluser_apiclient, None, None, None) | |
| vpc4 = self.create_vpc_for_account(self.domainapiclient, None, None, self.project) | |
| # Create VPC tier | |
| vpc1_tier1 = self.create_vpc_tier_for_account(self.apiclient, vpc1) | |
| vpc2_tier1 = self.create_vpc_tier_for_account(self.domainapiclient, vpc2) | |
| vpc3_tier1 = self.create_vpc_tier_for_account(self.normaluser_apiclient, vpc3) | |
| vpc4_tier1 = self.create_vpc_tier_for_account(self.domainapiclient, vpc4, self.project) | |
| # Create VPC private gateway with associated network (can be done by ROOT admin / domain admin / normal user) | |
| private_gateway1 = self.create_vpc_private_gateway(self.apiclient, vpc1, None, isolated_network1, True) | |
| private_gateway2 = self.create_vpc_private_gateway(self.domainapiclient, vpc2, None, isolated_network2, True) | |
| private_gateway3 = self.create_vpc_private_gateway(self.normaluser_apiclient, vpc3, None, isolated_network3, True) | |
| private_gateway4 = self.create_vpc_private_gateway(self.domainapiclient, vpc4, None, isolated_network4, True) | |
| # Check state of isolated networks (should be Implemented) | |
| self.check_network_state(self.apiclient, isolated_network1, None, NETWORK_STATE_IMPLEMENTED) | |
| self.check_network_state(self.domainapiclient, isolated_network2, None, NETWORK_STATE_IMPLEMENTED) | |
| self.check_network_state(self.normaluser_apiclient, isolated_network3, None, NETWORK_STATE_IMPLEMENTED) | |
| self.check_network_state(self.domainapiclient, isolated_network4, self.project, NETWORK_STATE_IMPLEMENTED) | |
| # Delete VPC private gateway of domain admin (should succeed) | |
| self.delete_vpc_private_gateway(self.domainapiclient, private_gateway2, True) | |
| # Wait for network GC to shut down the isolated networks | |
| gc_wait = Configurations.list(self.apiclient, name="network.gc.wait") | |
| gc_interval = Configurations.list(self.apiclient, name="network.gc.interval") | |
| total_sleep = 360 | |
| if gc_wait and gc_interval: | |
| self.logger.debug("network.gc.wait is ==> %s", gc_wait[0].value) | |
| self.logger.debug("network.gc.interval is ==> %s", gc_interval[0].value) | |
| total_sleep = max(int(gc_wait[0].value), int(gc_interval[0].value)) * 2 + 60 | |
| else: | |
| self.logger.debug("Could not retrieve the keys 'network.gc.interval' and 'network.gc.wait'. Sleeping for 6 minutes.") | |
| time.sleep(total_sleep) | |
| # Check state of isolated networks (1 should be Allocated, 2 should still be Implemented) | |
| self.check_network_state(self.apiclient, isolated_network1, None, NETWORK_STATE_IMPLEMENTED) | |
| self.check_network_state(self.domainapiclient, isolated_network2, None, NETWORK_STATE_ALLOCATED) | |
| self.check_network_state(self.normaluser_apiclient, isolated_network3, None, NETWORK_STATE_IMPLEMENTED) | |
| self.check_network_state(self.domainapiclient, isolated_network4, self.project, NETWORK_STATE_IMPLEMENTED) | |
| # Delete 2 VPC private gateway (should succeed) | |
| self.delete_vpc_private_gateway(self.apiclient, private_gateway1, True) | |
| self.delete_vpc_private_gateway(self.apiclient, private_gateway3, True) | |
| self.delete_vpc_private_gateway(self.apiclient, private_gateway4, True) | |
| # Delete VPC tiers (should succeed) | |
| self.delete_network(vpc1_tier1, self.apiclient, True) | |
| self.delete_network(vpc2_tier1, self.domainapiclient, True) | |
| self.delete_network(vpc3_tier1, self.normaluser_apiclient, True) | |
| self.delete_network(vpc4_tier1, self.domainapiclient, True) | |
| # Delete VPC | |
| self.delete_vpc(self.apiclient, vpc1) | |
| self.delete_vpc(self.domainapiclient, vpc2) | |
| self.delete_vpc(self.normaluser_apiclient, vpc3) | |
| self.delete_vpc(self.domainapiclient, vpc4) | |
| # Delete isolated networks (should succeed) | |
| self.delete_network(isolated_network1, self.apiclient, True) | |
| self.delete_network(isolated_network2, self.domainapiclient, True) | |
| self.delete_network(isolated_network3, self.normaluser_apiclient, True) | |
| self.delete_network(isolated_network4, self.domainapiclient, True) |