| # 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-shared networks |
| """ |
| |
| 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) |
| |
| 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 TestUserSharedNetworks(cloudstackTestCase): |
| """ |
| Test user-shared networks |
| """ |
| @classmethod |
| def setUpClass(cls): |
| cls.testClient = super( |
| TestUserSharedNetworks, |
| cls).getClsTestClient() |
| 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("TestUserSharedNetworks") |
| 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 user-shared networks (specifyVlan=true) |
| cls.network_offering_withvlan = NetworkOffering.create( |
| cls.apiclient, |
| cls.services["network_offering_shared"] |
| ) |
| cls.network_offering_withvlan.update(cls.apiclient, state='Enabled') |
| cls._cleanup.append(cls.network_offering_withvlan) |
| |
| # Create network offering for user-shared networks (specifyVlan=false) |
| cls.services["network_offering_shared"]["specifyVlan"] = "False" |
| cls.network_offering_novlan = NetworkOffering.create( |
| cls.apiclient, |
| cls.services["network_offering_shared"] |
| ) |
| cls.network_offering_novlan.update(cls.apiclient, state='Enabled') |
| cls._cleanup.append(cls.network_offering_novlan) |
| |
| # 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(TestUserSharedNetworks, cls).tearDownClass() |
| |
| def setUp(self): |
| self.cleanup = [] |
| |
| def tearDown(self): |
| super(TestUserSharedNetworks, self).tearDown() |
| |
| def create_shared_network_for_account(self, apiclient, domain, account, expected=True): |
| return self.create_shared_network_with_associated_network(apiclient, domain, account, None, None, expected) |
| |
| def create_shared_network_with_associated_network_for_domain(self, apiclient, domain, associated_network, expected=True): |
| return self.create_shared_network_with_associated_network(apiclient, domain, None, None, associated_network, expected) |
| |
| def create_shared_network_with_associated_network_for_caller(self, apiclient, project, associated_network, expected=True): |
| return self.create_shared_network_with_associated_network(apiclient, None, None, project, associated_network, expected) |
| |
| def create_shared_network_with_associated_network(self, apiclient, domain, account, project, associated_network, expected=True): |
| self.services["network2"]["acltype"] = "Account" |
| self.services["network2"]["name"] = "Test Network Shared - " + random_gen() |
| domain_id = None |
| account_name = None |
| project_id = None |
| if domain: |
| self.services["network2"]["acltype"] = "Domain" |
| domain_id = domain.id |
| if account: |
| self.services["network2"]["acltype"] = "Account" |
| account_name = account.name |
| if project: |
| self.services["network2"]["acltype"] = "Account" |
| project_id = project.id |
| associated_network_id = None |
| if associated_network: |
| associated_network_id = associated_network.id |
| try: |
| network = Network.create( |
| apiclient, |
| self.services["network2"], |
| domainid=domain_id, |
| accountid=account_name, |
| projectid=project_id, |
| associatednetworkid=associated_network_id, |
| zoneid=self.zone.id |
| ) |
| except Exception as ex: |
| network = None |
| if expected: |
| self.fail(f"Failed to create Shared network, but expected to succeed : {ex}") |
| if network and not expected: |
| self.fail("Shared network is created successfully, but expected to fail") |
| return network |
| |
| 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"]["displaytext"] = self.services["vpc"]["name"] |
| 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 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_user_shared_network_without_vlan(self): |
| """ Create user-shared networks without vlan """ |
| self.services["network2"]["networkoffering"] = self.network_offering_novlan.id |
| self.services["network2"]["vlan"] = None |
| |
| network1 = self.create_shared_network_for_account(self.apiclient, None, None) |
| network2 = self.create_shared_network_for_account(self.domainapiclient, None, None) |
| network3 = self.create_shared_network_for_account(self.normaluser_apiclient, None, None) |
| |
| self.delete_network(network1, self.apiclient) |
| self.delete_network(network2, self.domainapiclient) |
| self.delete_network(network3, self.normaluser_apiclient) |
| |
| network4 = self.create_shared_network_for_account(self.domainapiclient, self.sub_domain, self.normal_user, True) |
| self.delete_network(network4, self.normaluser_apiclient) |
| |
| network5 = self.create_shared_network_for_account(self.apiclient, self.sub_domain, None, True) |
| self.delete_network(network5, self.apiclient) |
| network6 = self.create_shared_network_for_account(self.domainapiclient, self.sub_domain, None, True) |
| self.delete_network(network6, self.apiclient) |
| self.create_shared_network_for_account(self.normaluser_apiclient, self.sub_domain, None, False) |
| |
| @attr(tags=["advanced"], required_hardware="false") |
| def test_02_create_shared_network_with_vlan(self): |
| """ Create shared networks with vlan |
| Only be created/deleted by root admin, Cannot create networks with same vlan |
| """ |
| self.services["network2"]["networkoffering"] = self.network_offering_withvlan.id |
| self.services["network2"]["vlan"] = 4000 |
| |
| self.create_shared_network_for_account(self.domainapiclient, self.sub_domain, None, False) |
| self.create_shared_network_for_account(self.normaluser_apiclient, self.sub_domain, None, False) |
| network1 = self.create_shared_network_for_account(self.apiclient, self.sub_domain, None, True) |
| self.create_shared_network_for_account(self.apiclient, self.sub_domain, self.normal_user, False) |
| self.delete_network(network1, self.domainapiclient, False) |
| self.delete_network(network1, self.apiclient, True) |
| |
| self.create_shared_network_for_account(self.domainapiclient, self.sub_domain, self.normal_user, False) |
| self.create_shared_network_for_account(self.normaluser_apiclient, self.sub_domain, self.normal_user, False) |
| network2 = self.create_shared_network_for_account(self.apiclient, self.sub_domain, self.normal_user, True) |
| self.delete_network(network2, self.domainapiclient, False) |
| self.delete_network(network2, self.normaluser_apiclient, False) |
| self.delete_network(network2, self.apiclient, True) |
| |
| @attr(tags=["advanced"], required_hardware="false") |
| def test_03_create_domain_shared_network_with_associated_network(self): |
| """ Create domain-level shared networks with associated network """ |
| |
| self.services["network2"]["networkoffering"] = self.network_offering_novlan.id |
| self.services["network2"]["vlan"] = None |
| |
| # Create isolated networks |
| isolated_network1 = self.create_isolated_network_for_account(self.apiclient, None, None, None, True) |
| isolated_network2 = self.create_isolated_network_for_account(self.apiclient, self.sub_domain, self.domain_admin, None, True) |
| isolated_network3 = self.create_isolated_network_for_account(self.apiclient, self.sub_domain, self.normal_user, None, True) |
| isolated_network4 = self.create_isolated_network_for_account(self.apiclient, self.sub_domain, None, self.project, True) |
| |
| # Create domain-level shared network with associated_network (caller must be root admin/domain admin, must be in same domain) |
| self.create_shared_network_with_associated_network_for_domain(self.apiclient, self.sub_domain, isolated_network1, False) |
| self.create_shared_network_with_associated_network_for_domain(self.domainapiclient, self.sub_domain, isolated_network1, False) |
| self.create_shared_network_with_associated_network_for_domain(self.normaluser_apiclient, self.sub_domain, isolated_network1, False) |
| self.create_shared_network_with_associated_network_for_domain(self.normaluser_apiclient, self.sub_domain, isolated_network2, False) |
| self.create_shared_network_with_associated_network_for_domain(self.normaluser_apiclient, self.sub_domain, isolated_network3, False) |
| self.create_shared_network_with_associated_network_for_domain(self.normaluser_apiclient, self.sub_domain, isolated_network4, False) |
| shared_network1 = self.create_shared_network_with_associated_network_for_domain(self.apiclient, self.domain, isolated_network1, True) |
| shared_network2 = self.create_shared_network_with_associated_network_for_domain(self.domainapiclient, self.sub_domain, isolated_network2, True) |
| shared_network3 = self.create_shared_network_with_associated_network_for_domain(self.domainapiclient, self.sub_domain, isolated_network3, True) |
| shared_network4 = self.create_shared_network_with_associated_network_for_domain(self.domainapiclient, self.sub_domain, 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 isolated networks (should fail) |
| self.delete_network(isolated_network1, self.apiclient, False) |
| self.delete_network(isolated_network2, self.apiclient, False) |
| self.delete_network(isolated_network3, self.apiclient, False) |
| self.delete_network(isolated_network4, self.apiclient, False) |
| |
| # 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 domain-level shared network with associated vpc tier (caller must be root admin/domain admin, must be in same domain) |
| self.create_shared_network_with_associated_network_for_domain(self.apiclient, self.sub_domain, vpc1_tier1, False) |
| self.create_shared_network_with_associated_network_for_domain(self.domainapiclient, self.sub_domain, vpc1_tier1, False) |
| self.create_shared_network_with_associated_network_for_domain(self.normaluser_apiclient, self.sub_domain, vpc1_tier1, False) |
| self.create_shared_network_with_associated_network_for_domain(self.normaluser_apiclient, self.sub_domain, vpc2_tier1, False) |
| self.create_shared_network_with_associated_network_for_domain(self.normaluser_apiclient, self.sub_domain, vpc3_tier1, False) |
| self.create_shared_network_with_associated_network_for_domain(self.normaluser_apiclient, self.sub_domain, vpc4_tier1, False) |
| shared_network_vpctier1 = self.create_shared_network_with_associated_network_for_domain(self.apiclient, self.domain, vpc1_tier1, True) |
| shared_network_vpctier2 = self.create_shared_network_with_associated_network_for_domain(self.domainapiclient, self.sub_domain, vpc2_tier1, True) |
| shared_network_vpctier3 = self.create_shared_network_with_associated_network_for_domain(self.domainapiclient, self.sub_domain, vpc3_tier1, True) |
| shared_network_vpctier4 = self.create_shared_network_with_associated_network_for_domain(self.domainapiclient, self.sub_domain, vpc4_tier1, True) |
| |
| # Check state of vpc tiers (should be Implemented) |
| self.check_network_state(self.apiclient, vpc1_tier1, None, NETWORK_STATE_IMPLEMENTED) |
| self.check_network_state(self.domainapiclient, vpc2_tier1, None, NETWORK_STATE_IMPLEMENTED) |
| self.check_network_state(self.normaluser_apiclient, vpc3_tier1, None, NETWORK_STATE_IMPLEMENTED) |
| self.check_network_state(self.domainapiclient, vpc4_tier1, self.project, NETWORK_STATE_IMPLEMENTED) |
| |
| # Delete vpc tiers(should fail) |
| self.delete_network(vpc1_tier1, self.apiclient, False) |
| self.delete_network(vpc2_tier1, self.apiclient, False) |
| self.delete_network(vpc3_tier1, self.apiclient, False) |
| self.delete_network(vpc4_tier1, self.apiclient, False) |
| |
| # Delete shared networks associated to domain admin's isolated network or vpc tier(should succeed) |
| self.delete_network(shared_network2, self.domainapiclient, True) |
| self.delete_network(shared_network_vpctier2, self.domainapiclient, 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, 3 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) |
| |
| self.check_network_state(self.apiclient, vpc1_tier1, None, NETWORK_STATE_IMPLEMENTED) |
| self.check_network_state(self.domainapiclient, vpc2_tier1, None, NETWORK_STATE_ALLOCATED) |
| self.check_network_state(self.normaluser_apiclient, vpc3_tier1, None, NETWORK_STATE_IMPLEMENTED) |
| self.check_network_state(self.domainapiclient, vpc4_tier1, self.project, NETWORK_STATE_IMPLEMENTED) |
| |
| # Check state of shared network (should be Setup) |
| self.check_network_state(self.apiclient, shared_network1, None, NETWORK_STATE_SETUP) |
| self.check_network_state(self.normaluser_apiclient, shared_network3, None, NETWORK_STATE_SETUP) |
| self.check_network_state(self.domainapiclient, shared_network4, None, NETWORK_STATE_SETUP) |
| |
| self.check_network_state(self.apiclient, shared_network_vpctier1, None, NETWORK_STATE_SETUP) |
| self.check_network_state(self.normaluser_apiclient, shared_network_vpctier3, None, NETWORK_STATE_SETUP) |
| self.check_network_state(self.domainapiclient, shared_network_vpctier4, None, NETWORK_STATE_SETUP) |
| |
| # Delete admin's shared networks (should succeed) |
| self.delete_network(shared_network1, self.apiclient, True) |
| self.delete_network(shared_network3, self.domainapiclient, True) |
| self.delete_network(shared_network4, self.domainapiclient, True) |
| |
| self.delete_network(shared_network_vpctier1, self.apiclient, True) |
| self.delete_network(shared_network_vpctier3, self.domainapiclient, True) |
| self.delete_network(shared_network_vpctier4, self.domainapiclient, True) |
| |
| # Delete admin's and domain admin's isolated network, but keep the normal user's network |
| # normal user's shared network and isolated network should be removed in tearDown successfully |
| self.delete_network(isolated_network1, self.apiclient, True) |
| self.delete_network(isolated_network2, self.domainapiclient, True) |
| |
| self.delete_network(vpc1_tier1, self.apiclient, True) |
| self.delete_vpc(self.apiclient, vpc1) |
| |
| @attr(tags=["advanced"], required_hardware="false") |
| def test_04_create_account_shared_network_with_associated_network(self): |
| """ Create account-level shared networks with associated network """ |
| |
| self.services["network2"]["networkoffering"] = self.network_offering_novlan.id |
| self.services["network2"]["vlan"] = None |
| |
| # Create isolated networks |
| isolated_network1 = self.create_isolated_network_for_account(self.apiclient, None, None, None, True) |
| isolated_network2 = self.create_isolated_network_for_account(self.apiclient, self.sub_domain, self.domain_admin, None, True) |
| isolated_network3 = self.create_isolated_network_for_account(self.apiclient, self.sub_domain, self.normal_user, None, True) |
| isolated_network4 = self.create_isolated_network_for_account(self.apiclient, self.sub_domain, None, self.project, True) |
| |
| # 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 shared networks with associated network (must be same owner) |
| self.create_shared_network_with_associated_network_for_caller(self.domainapiclient, None, isolated_network1, False) |
| self.create_shared_network_with_associated_network_for_caller(self.domainapiclient, None, isolated_network3, False) |
| self.create_shared_network_with_associated_network_for_caller(self.domainapiclient, None, isolated_network4, False) |
| self.create_shared_network_with_associated_network_for_caller(self.normaluser_apiclient, None, isolated_network1, False) |
| self.create_shared_network_with_associated_network_for_caller(self.normaluser_apiclient, None, isolated_network2, False) |
| self.create_shared_network_with_associated_network_for_caller(self.normaluser_apiclient, None, isolated_network4, False) |
| shared_network1 = self.create_shared_network_with_associated_network_for_caller(self.apiclient, None, isolated_network1, True) |
| shared_network2 = self.create_shared_network_with_associated_network_for_caller(self.domainapiclient, None, isolated_network2, True) |
| shared_network3 = self.create_shared_network_with_associated_network_for_caller(self.normaluser_apiclient, None, isolated_network3, True) |
| shared_network4 = self.create_shared_network_with_associated_network_for_caller(self.domainapiclient, self.project, 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 isolated networks (should fail) |
| self.delete_network(isolated_network1, self.apiclient, False) |
| self.delete_network(isolated_network2, self.apiclient, False) |
| self.delete_network(isolated_network3, self.apiclient, False) |
| self.delete_network(isolated_network4, self.apiclient, False) |
| |
| # 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 account-level shared network with associated vpc tier (caller must be root admin/domain admin, must be in same domain) |
| self.create_shared_network_with_associated_network_for_caller(self.domainapiclient, None, vpc1_tier1, False) |
| self.create_shared_network_with_associated_network_for_caller(self.domainapiclient, None, vpc3_tier1, False) |
| self.create_shared_network_with_associated_network_for_caller(self.domainapiclient, None, vpc4_tier1, False) |
| self.create_shared_network_with_associated_network_for_caller(self.normaluser_apiclient, None, vpc1_tier1, False) |
| self.create_shared_network_with_associated_network_for_caller(self.normaluser_apiclient, None, vpc2_tier1, False) |
| self.create_shared_network_with_associated_network_for_caller(self.normaluser_apiclient, None, vpc4_tier1, False) |
| shared_network_vpctier1 = self.create_shared_network_with_associated_network_for_caller(self.apiclient, None, vpc1_tier1, True) |
| shared_network_vpctier2 = self.create_shared_network_with_associated_network_for_caller(self.domainapiclient, None, vpc2_tier1, True) |
| shared_network_vpctier3 = self.create_shared_network_with_associated_network_for_caller(self.normaluser_apiclient, None, vpc3_tier1, True) |
| shared_network_vpctier4 = self.create_shared_network_with_associated_network_for_caller(self.domainapiclient, self.project, vpc4_tier1, True) |
| |
| # Check state of vpc tiers (should be Implemented) |
| self.check_network_state(self.apiclient, vpc1_tier1, None, NETWORK_STATE_IMPLEMENTED) |
| self.check_network_state(self.domainapiclient, vpc2_tier1, None, NETWORK_STATE_IMPLEMENTED) |
| self.check_network_state(self.normaluser_apiclient, vpc3_tier1, None, NETWORK_STATE_IMPLEMENTED) |
| self.check_network_state(self.domainapiclient, vpc4_tier1, self.project, NETWORK_STATE_IMPLEMENTED) |
| |
| # Delete vpc tiers(should fail) |
| self.delete_network(vpc1_tier1, self.apiclient, False) |
| self.delete_network(vpc2_tier1, self.apiclient, False) |
| self.delete_network(vpc3_tier1, self.apiclient, False) |
| self.delete_network(vpc4_tier1, self.apiclient, False) |
| |
| # Delete shared networks associated to domain admin's isolated network or vpc tier(should succeed) |
| self.delete_network(shared_network2, self.domainapiclient, True) |
| self.delete_network(shared_network_vpctier2, self.domainapiclient, 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, 3 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) |
| |
| self.check_network_state(self.apiclient, vpc1_tier1, None, NETWORK_STATE_IMPLEMENTED) |
| self.check_network_state(self.domainapiclient, vpc2_tier1, None, NETWORK_STATE_ALLOCATED) |
| self.check_network_state(self.normaluser_apiclient, vpc3_tier1, None, NETWORK_STATE_IMPLEMENTED) |
| self.check_network_state(self.domainapiclient, vpc4_tier1, self.project, NETWORK_STATE_IMPLEMENTED) |
| |
| # Check state of shared network (should be Setup) |
| self.check_network_state(self.apiclient, shared_network1, None, NETWORK_STATE_SETUP) |
| self.check_network_state(self.normaluser_apiclient, shared_network3, None, NETWORK_STATE_SETUP) |
| self.check_network_state(self.domainapiclient, shared_network4, self.project, NETWORK_STATE_SETUP) |
| |
| self.check_network_state(self.apiclient, shared_network_vpctier1, None, NETWORK_STATE_SETUP) |
| self.check_network_state(self.normaluser_apiclient, shared_network_vpctier3, None, NETWORK_STATE_SETUP) |
| self.check_network_state(self.domainapiclient, shared_network_vpctier4, self.project, NETWORK_STATE_SETUP) |
| |
| # Delete admin's shared networks (should succeed) |
| self.delete_network(shared_network1, self.apiclient, True) |
| self.delete_network(shared_network_vpctier1, self.apiclient, True) |
| |
| # Delete admin's and domain admin's isolated network, but keep the normal user's network |
| # normal user's shared network and isolated network should be removed in tearDown successfully |
| self.delete_network(isolated_network1, self.apiclient, True) |
| self.delete_network(isolated_network2, self.domainapiclient, True) |
| |
| self.delete_network(vpc1_tier1, self.apiclient, True) |
| self.delete_vpc(self.apiclient, vpc1) |
| self.delete_network(vpc2_tier1, self.domainapiclient, True) |
| self.delete_vpc(self.domainapiclient, vpc2) |
