DaanHoogland commented on a change in pull request #5769:
URL: https://github.com/apache/cloudstack/pull/5769#discussion_r778020860
##########
File path: api/src/main/java/com/cloud/network/GuestVlan.java
##########
@@ -16,17 +16,26 @@
// under the License.
package com.cloud.network;
-import org.apache.cloudstack.api.Identity;
import org.apache.cloudstack.api.InternalIdentity;
-public interface GuestVlan extends InternalIdentity, Identity {
+import java.util.Date;
+
+public interface GuestVlan extends InternalIdentity {
@Override
public long getId();
- public long getAccountId();
+ Date getTakenAt();
+
+ String getVnet();
+
+ String getReservationId();
+
+ Long getAccountId();
+
+ long getDataCenterId();
- public String getGuestVlanRange();
+ long getPhysicalNetworkId();
- public long getPhysicalNetworkId();
+ Long getAccountGuestVlanMapId();
Review comment:
why the discrepencies between `long` and `Long`, @weizhouapache ? Are
both `accountId` and `accountGuestVlanMapId` optional? If so can you add
javadoc?
##########
File path: server/src/main/java/com/cloud/api/ApiResponseHelper.java
##########
@@ -2424,6 +2431,17 @@ public NetworkResponse
createNetworkResponse(ResponseView view, Network network)
response.setVpcName(vpc.getName());
}
}
+
+ final NetworkDetailVO detail =
networkDetailsDao.findDetail(network.getId(), Network.AssociatedNetworkId);
+ if (detail != null) {
+ Long associatedNetworkId = Long.valueOf(detail.getValue());
+ NetworkVO associatedNetwork =
ApiDBUtils.findNetworkById(associatedNetworkId);
+ if (associatedNetwork != null) {
+ response.setAssociatedNetworkId(associatedNetwork.getUuid());
+ response.setAssociatedNetworkName(associatedNetwork.getName());
+ }
+ }
Review comment:
can you extract `addAssociatedNetworkToResponse(NetworkResponse
response)`, please?
##########
File path: test/integration/smoke/test_network_permissions.py
##########
@@ -0,0 +1,757 @@
+# 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 network permissions
+"""
+
+import logging
+
+from nose.plugins.attrib import attr
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+from marvin.lib.base import (Account,
+ Configurations,
+ Domain,
+ Project,
+ ServiceOffering,
+ VirtualMachine,
+ Zone,
+ Network,
+ NetworkOffering,
+ NetworkPermission,
+ NIC,
+ PublicIPAddress,
+ LoadBalancerRule,
+ NATRule,
+ StaticNATRule,
+
+ SSHKeyPair)
+
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template)
+
+NETWORK_FILTER_ACCOUNT = 'account'
+NETWORK_FILTER_DOMAIN = 'domain'
+NETWORK_FILTER_ACCOUNT_DOMAIN = 'accountdomain'
+NETWORK_FILTER_SHARED = 'shared'
+NETWORK_FILTER_ALL = 'all'
+
+class TestNetworkPermissions(cloudstackTestCase):
+ """
+ Test user-shared networks
+ """
+ @classmethod
+ def setUpClass(cls):
+ cls.testClient = super(
+ TestNetworkPermissions,
+ 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("TestNetworkPermissions")
+ 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["isolated_network_offering"]
+ )
+ cls.network_offering_isolated.update(cls.apiclient, state='Enabled')
+ cls._cleanup.append(cls.network_offering_isolated)
+
+ # 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.network_owner = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11A"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.network_owner)
+
+ cls.other_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11B"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.other_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.domainadmin_apiclient = cls.testClient.getUserApiClient(
+ cls.domainadmin_user.username, cls.sub_domain.name
+ )
+ cls.networkowner_user = cls.network_owner.user[0]
+ cls.user_apiclient = cls.testClient.getUserApiClient(
+ cls.networkowner_user.username, cls.sub_domain.name
+ )
+
+ cls.otheruser_user = cls.other_user.user[0]
+ cls.otheruser_apiclient = cls.testClient.getUserApiClient(
+ cls.otheruser_user.username, cls.sub_domain.name
+ )
+
+ # Create networks for domain admin, normal user and project
+ cls.services["network"]["name"] = "Test Network Isolated - Project"
+ cls.project_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ projectid=cls.project.id,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Domain
admin"
+ cls.domainadmin_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.domain_admin.name,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Normal user"
+ cls.user_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.network_owner.name,
+ zoneid=cls.zone.id
+ )
Review comment:
should this be cleaned (added to `cls._cleanup`)
##########
File path: test/integration/smoke/test_network_permissions.py
##########
@@ -0,0 +1,757 @@
+# 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 network permissions
+"""
+
+import logging
+
+from nose.plugins.attrib import attr
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+from marvin.lib.base import (Account,
+ Configurations,
+ Domain,
+ Project,
+ ServiceOffering,
+ VirtualMachine,
+ Zone,
+ Network,
+ NetworkOffering,
+ NetworkPermission,
+ NIC,
+ PublicIPAddress,
+ LoadBalancerRule,
+ NATRule,
+ StaticNATRule,
+
+ SSHKeyPair)
+
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template)
+
+NETWORK_FILTER_ACCOUNT = 'account'
+NETWORK_FILTER_DOMAIN = 'domain'
+NETWORK_FILTER_ACCOUNT_DOMAIN = 'accountdomain'
+NETWORK_FILTER_SHARED = 'shared'
+NETWORK_FILTER_ALL = 'all'
+
+class TestNetworkPermissions(cloudstackTestCase):
+ """
+ Test user-shared networks
+ """
+ @classmethod
+ def setUpClass(cls):
+ cls.testClient = super(
+ TestNetworkPermissions,
+ 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("TestNetworkPermissions")
+ 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["isolated_network_offering"]
+ )
+ cls.network_offering_isolated.update(cls.apiclient, state='Enabled')
+ cls._cleanup.append(cls.network_offering_isolated)
+
+ # 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.network_owner = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11A"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.network_owner)
+
+ cls.other_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11B"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.other_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.domainadmin_apiclient = cls.testClient.getUserApiClient(
+ cls.domainadmin_user.username, cls.sub_domain.name
+ )
+ cls.networkowner_user = cls.network_owner.user[0]
+ cls.user_apiclient = cls.testClient.getUserApiClient(
+ cls.networkowner_user.username, cls.sub_domain.name
+ )
+
+ cls.otheruser_user = cls.other_user.user[0]
+ cls.otheruser_apiclient = cls.testClient.getUserApiClient(
+ cls.otheruser_user.username, cls.sub_domain.name
+ )
+
+ # Create networks for domain admin, normal user and project
+ cls.services["network"]["name"] = "Test Network Isolated - Project"
+ cls.project_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ projectid=cls.project.id,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Domain
admin"
+ cls.domainadmin_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.domain_admin.name,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Normal user"
+ cls.user_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.network_owner.name,
+ zoneid=cls.zone.id
+ )
+
+ @classmethod
+ def tearDownClass(cls):
+ super().tearDownClass()
Review comment:
```suggestion
super(TestNetworkPermissions, cls).tearDownClass()
```
I don't think it will work otherwise?
##########
File path: test/integration/smoke/test_network_permissions.py
##########
@@ -0,0 +1,757 @@
+# 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 network permissions
+"""
+
+import logging
+
+from nose.plugins.attrib import attr
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+from marvin.lib.base import (Account,
+ Configurations,
+ Domain,
+ Project,
+ ServiceOffering,
+ VirtualMachine,
+ Zone,
+ Network,
+ NetworkOffering,
+ NetworkPermission,
+ NIC,
+ PublicIPAddress,
+ LoadBalancerRule,
+ NATRule,
+ StaticNATRule,
+
+ SSHKeyPair)
+
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template)
+
+NETWORK_FILTER_ACCOUNT = 'account'
+NETWORK_FILTER_DOMAIN = 'domain'
+NETWORK_FILTER_ACCOUNT_DOMAIN = 'accountdomain'
+NETWORK_FILTER_SHARED = 'shared'
+NETWORK_FILTER_ALL = 'all'
+
+class TestNetworkPermissions(cloudstackTestCase):
+ """
+ Test user-shared networks
+ """
+ @classmethod
+ def setUpClass(cls):
+ cls.testClient = super(
+ TestNetworkPermissions,
+ 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("TestNetworkPermissions")
+ 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["isolated_network_offering"]
+ )
+ cls.network_offering_isolated.update(cls.apiclient, state='Enabled')
+ cls._cleanup.append(cls.network_offering_isolated)
+
+ # 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.network_owner = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11A"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.network_owner)
+
+ cls.other_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11B"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.other_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.domainadmin_apiclient = cls.testClient.getUserApiClient(
+ cls.domainadmin_user.username, cls.sub_domain.name
+ )
+ cls.networkowner_user = cls.network_owner.user[0]
+ cls.user_apiclient = cls.testClient.getUserApiClient(
+ cls.networkowner_user.username, cls.sub_domain.name
+ )
+
+ cls.otheruser_user = cls.other_user.user[0]
+ cls.otheruser_apiclient = cls.testClient.getUserApiClient(
+ cls.otheruser_user.username, cls.sub_domain.name
+ )
+
+ # Create networks for domain admin, normal user and project
+ cls.services["network"]["name"] = "Test Network Isolated - Project"
+ cls.project_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ projectid=cls.project.id,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Domain
admin"
+ cls.domainadmin_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.domain_admin.name,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Normal user"
+ cls.user_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.network_owner.name,
+ zoneid=cls.zone.id
+ )
+
+ @classmethod
+ def tearDownClass(cls):
+ super().tearDownClass()
+
+ def setUp(self):
+ self.cleanup = []
+ self.virtual_machine = None
+
+ def tearDown(self):
+ super().tearDown()
+
+ def list_network(self, apiclient, account, network, project,
network_filter=None, expected=True):
+ # List networks by apiclient, account, network, project and network
network_filter
+ # If account is specified, list the networks which can be used by the
domain (canusefordeploy=true,listall=false)
+ # otherwise canusefordeploy is None and listall is True.
+ domain_id = None
+ account_name = None
+ project_id = None
+ canusefordeploy = None
+ list_all = True
+ if account:
+ domain_id = account.domainid
+ account_name = account.name
+ canusefordeploy = True
+ list_all = False
+ if project:
+ project_id = project.id
+ networks = None
+ try:
+ networks = Network.list(
+ apiclient,
+ canusefordeploy=canusefordeploy,
+ listall=list_all,
+ networkfilter= network_filter,
+ domainid=domain_id,
+ account=account_name,
+ projectid=project_id,
+ id=network.id
+ )
+ if isinstance(networks, list) and len(networks) > 0:
+ if not expected:
+ self.fail("Found the network, but expected to fail")
+ elif expected:
+ self.fail("Failed to find the network, but expected to
succeed")
+ except Exception as ex:
+ networks = None
+ if expected:
+ self.fail(f"Failed to list network, but expected to succeed :
{ex}")
+ if networks and not expected:
+ self.fail("network is listed successfully, but expected to fail")
+
+ def list_network_by_filters(self, apiclient, account, network, project,
expected_results=None):
+ # expected results in order: account/domain/accountdomain/shared/all
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ACCOUNT, expected_results[0])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_DOMAIN, expected_results[1])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ACCOUNT_DOMAIN, expected_results[2])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_SHARED, expected_results[3])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ALL, expected_results[4])
+
+ def create_network_permission(self, apiclient, network, account, project,
expected=True):
+ account_id = None
+ project_id = None
+ if account:
+ account_id = account.id
+ if project:
+ project_id = project.id
+ result = True
+ try:
+ NetworkPermission.create(
+ apiclient,
+ networkid=network.id,
+ accountids=account_id,
+ projectids=project_id
+ )
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to create network permissions, but expected
to succeed : {ex}")
+ if result and not expected:
+ self.fail("network permission is created successfully, but
expected to fail")
+
+ def remove_network_permission(self, apiclient, network, account, project,
expected=True):
+ account_id = None
+ project_id = None
+ if account:
+ account_id = account.id
+ if project:
+ project_id = project.id
+ result = True
+ try:
+ NetworkPermission.remove(
+ apiclient,
+ networkid=network.id,
+ accountids=account_id,
+ projectids=project_id
+ )
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to remove network permissions, but expected
to succeed : {ex}")
+ if result and not expected:
+ self.fail("network permission is removed successfully, but
expected to fail")
+
+ def reset_network_permission(self, apiclient, network, expected=True):
+ result = True
+ try:
+ NetworkPermission.reset(
+ apiclient,
+ networkid=network.id
+ )
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to reset network permissions, but expected
to succeed : {ex}")
+ if result and not expected:
+ self.fail("network permission is reset successfully, but expected
to fail")
+
+ def exec_command(self, apiclient_str, command, expected=None):
+ result = True
+ try:
+ command = command.format(apiclient = apiclient_str)
+ exec(command)
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to execute command '{command}' with
exception : {ex}")
+ if result and expected is False:
+ self.fail(f"command {command} is executed successfully, but
expected to fail")
+ if expected is None:
+ # if expected is None, display the command and result
+ self.logger.info(f"Result of command '{command}' : {result}")
+ return result
+
+ @attr(tags=["advanced"], required_hardware="false")
+ def test_01_network_permission_on_project_network(self):
+ """ Testing network permissions on project network """
+
+ self.create_network_permission(self.apiclient, self.project_network,
self.domain_admin, None, expected=False)
+ self.create_network_permission(self.domainadmin_apiclient,
self.project_network, self.domain_admin, None, expected=False)
+ self.create_network_permission(self.user_apiclient,
self.project_network, self.network_owner, None, expected=False)
+
+ @attr(tags=["advanced"], required_hardware="false")
+ def test_02_network_permission_on_user_network(self):
+ """ Testing network permissions on user network """
+
+ # List user network by domain admin
+ self.list_network_by_filters(self.domainadmin_apiclient, None,
self.user_network, None, [True, False, True, False, True])
+ self.list_network_by_filters(self.domainadmin_apiclient,
self.domain_admin, self.user_network, None, [False, False, False, False, False])
+
+ # Create network permissions
+ self.create_network_permission(self.apiclient, self.user_network,
self.domain_admin, None, expected=True)
+ self.create_network_permission(self.domainadmin_apiclient,
self.user_network, self.domain_admin, None, expected=True)
+ self.create_network_permission(self.user_apiclient, self.user_network,
self.network_owner, None, expected=True)
+ self.create_network_permission(self.user_apiclient, self.user_network,
self.other_user, None, expected=True)
+ self.create_network_permission(self.user_apiclient, self.user_network,
None, self.project, expected=False)
+ self.create_network_permission(self.domainadmin_apiclient,
self.user_network, None, self.project, expected=True)
+ self.create_network_permission(self.otheruser_apiclient,
self.user_network, self.network_owner, None, expected=False)
+
+ # List domain admin network by domain admin
+ self.list_network_by_filters(self.domainadmin_apiclient, None,
self.domainadmin_network, None, [True, False, True, False, True])
+ self.list_network_by_filters(self.domainadmin_apiclient,
self.domain_admin, self.domainadmin_network, None, [True, False, True, False,
True])
+ # List user network by domain admin
+ self.list_network_by_filters(self.domainadmin_apiclient, None,
self.user_network, None, [True, False, True, True, True])
+ self.list_network_by_filters(self.domainadmin_apiclient,
self.domain_admin, self.user_network, None, [False, False, False, True, True])
+ # List user network by user
+ self.list_network_by_filters(self.user_apiclient, None,
self.user_network, None, [True, False, True, False, True])
+ self.list_network_by_filters(self.user_apiclient, self.network_owner,
self.user_network, None, [True, False, True, False, True])
+ # List user network by other user
+ self.list_network_by_filters(self.otheruser_apiclient, None,
self.user_network, None, [False, False, False, True, True])
+ self.list_network_by_filters(self.otheruser_apiclient,
self.network_owner, self.user_network, None, [False, False, False, False,
False])
+
+ # Remove network permissions
+ self.remove_network_permission(self.domainadmin_apiclient,
self.user_network, self.domain_admin, None, expected=True)
+ # List user network by domain admin
+ self.list_network_by_filters(self.domainadmin_apiclient, None,
self.user_network, None, [True, False, True, True, True])
+ self.list_network_by_filters(self.domainadmin_apiclient,
self.domain_admin, self.user_network, None, [False, False, False, False, False])
+
+ # Reset network permissions
+ self.reset_network_permission(self.domainadmin_apiclient,
self.user_network, expected=True)
+ # List user network by domain admin
+ self.list_network_by_filters(self.domainadmin_apiclient, None,
self.user_network, None, [True, False, True, False, True])
+ self.list_network_by_filters(self.domainadmin_apiclient,
self.domain_admin, self.user_network, None, [False, False, False, False, False])
+
+ @attr(tags=["advanced"], required_hardware="false")
+ def test_03_network_operations_on_created_vm_of_otheruser(self):
+ """ Testing network operations on a create vm owned by other user"""
+
+ # 1. Create an Isolated network by other user
+ self.services["network"]["name"] = "Test Network Isolated - Other user"
+ otheruser_network = Network.create(
+ self.otheruser_apiclient,
+ self.services["network"],
+ networkofferingid=self.network_offering_isolated.id,
+ zoneid=self.zone.id
+ )
+ self.cleanup = [otheruser_network]
+
+ # 2. Deploy vm1 on other user's network
+ self.virtual_machine = VirtualMachine.create(
+ self.otheruser_apiclient,
+ self.services["virtual_machine"],
+ templateid=self.template.id,
+ serviceofferingid=self.service_offering.id,
+ networkids=otheruser_network.id,
+ zoneid=self.zone.id
+ )
Review comment:
```suggestion
)
self.cleanup.append(self.virtual_machine)
```
##########
File path: server/src/main/java/com/cloud/network/NetworkServiceImpl.java
##########
@@ -4560,13 +4783,22 @@ public Network createPrivateNetwork(final String
networkName, final String displ
final String cidr = NetUtils.ipAndNetMaskToCidr(gateway, netmask);
- URI uri = BroadcastDomainType.fromString(broadcastUriString);
- final String uriString = uri.toString();
- BroadcastDomainType tiep = BroadcastDomainType.getSchemeValue(uri);
- // numeric vlan or vlan uri are ok for now
- // TODO make a test for any supported scheme
- if (!(tiep == BroadcastDomainType.Vlan || tiep ==
BroadcastDomainType.Lswitch)) {
- throw new InvalidParameterValueException("unsupported type of
broadcastUri specified: " + broadcastUriString);
+ final String uriString;
+ if (broadcastUriString != null) {
+ URI uri = BroadcastDomainType.fromString(broadcastUriString);
+ uriString = uri.toString();
+ BroadcastDomainType tiep = BroadcastDomainType.getSchemeValue(uri);
+ // numeric vlan or vlan uri are ok for now
+ // TODO make a test for any supported scheme
+ if (!(tiep == BroadcastDomainType.Vlan || tiep ==
BroadcastDomainType.Lswitch)) {
+ throw new InvalidParameterValueException("unsupported type of
broadcastUri specified: " + broadcastUriString);
+ }
+ } else if (associatedNetworkId != null) {
+ DataCenter zone = _dcDao.findById(pNtwk.getDataCenterId());
+ Network associatedNetwork =
implementAssociatedNetwork(associatedNetworkId, caller, owner, zone, null,
owner.getAccountId(), cidr, startIp, endIp);
+ uriString = associatedNetwork.getBroadcastUri().toString();
+ } else {
+ throw new InvalidParameterValueException("One of uri and
associatedNetworkId must be passed");
Review comment:
extra method for this block?
##########
File path: test/integration/smoke/test_user_shared_network.py
##########
@@ -0,0 +1,630 @@
+# 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 sub-domain
+ cls.sub_domain = Domain.create(
+ cls.apiclient,
+ cls.services["acl"]["domain1"]
+ )
+
+ # 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.normal_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD1B"],
+ domainid=cls.sub_domain.id
+ )
+ # Create project
+ cls.project = Project.create(
+ cls.apiclient,
+ cls.services["project"],
+ account=cls.domain_admin.name,
+ domainid=cls.domain_admin.domainid
+ )
+
+ # Create small service offering
+ cls.service_offering = ServiceOffering.create(
+ cls.apiclient,
+ cls.services["service_offerings"]["small"]
+ )
+
+ # 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')
+
+ # 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')
+
+ # 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')
+
+ # Create vpc offering
+ cls.vpc_offering = VpcOffering.create(
+ cls.apiclient,
+ cls.services["vpc_offering_multi_lb"])
+ cls.vpc_offering.update(cls.apiclient, state='Enabled')
+
+ # 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')
+
+ # 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
+ )
+
+ cls._cleanup.append(cls.service_offering)
+ cls._cleanup.append(cls.network_offering_withvlan)
+ cls._cleanup.append(cls.network_offering_novlan)
+ cls._cleanup.append(cls.network_offering_isolated)
+ cls._cleanup.append(cls.network_offering_vpc)
+ cls._cleanup.append(cls.vpc_offering)
+
+ cls._cleanup.append(cls.sub_domain)
+ cls._cleanup.append(cls.normal_user)
+ cls._cleanup.append(cls.domain_admin)
+ cls._cleanup.append(cls.project)
Review comment:
I would put these directly after creation. In case something goes wrong
in the constructor we are still stuck with left over resources otherwise. In
general I even add things to the cleanup lists even if I delete them myself
just to remove them from the list if I do, to make sure the test is runnable in
production environments.
##########
File path: test/integration/smoke/test_user_shared_network.py
##########
@@ -0,0 +1,630 @@
+# 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 sub-domain
+ cls.sub_domain = Domain.create(
+ cls.apiclient,
+ cls.services["acl"]["domain1"]
+ )
+
+ # 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.normal_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD1B"],
+ domainid=cls.sub_domain.id
+ )
+ # Create project
+ cls.project = Project.create(
+ cls.apiclient,
+ cls.services["project"],
+ account=cls.domain_admin.name,
+ domainid=cls.domain_admin.domainid
+ )
+
+ # Create small service offering
+ cls.service_offering = ServiceOffering.create(
+ cls.apiclient,
+ cls.services["service_offerings"]["small"]
+ )
+
+ # 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')
+
+ # 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')
+
+ # 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')
+
+ # Create vpc offering
+ cls.vpc_offering = VpcOffering.create(
+ cls.apiclient,
+ cls.services["vpc_offering_multi_lb"])
+ cls.vpc_offering.update(cls.apiclient, state='Enabled')
+
+ # 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')
+
+ # 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
+ )
+
+ cls._cleanup.append(cls.service_offering)
+ cls._cleanup.append(cls.network_offering_withvlan)
+ cls._cleanup.append(cls.network_offering_novlan)
+ cls._cleanup.append(cls.network_offering_isolated)
+ cls._cleanup.append(cls.network_offering_vpc)
+ cls._cleanup.append(cls.vpc_offering)
+
+ cls._cleanup.append(cls.sub_domain)
+ cls._cleanup.append(cls.normal_user)
+ cls._cleanup.append(cls.domain_admin)
+ cls._cleanup.append(cls.project)
+
+ @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,
+ )
Review comment:
here `network` should be removed from `self.cleanup` again
##########
File path: test/integration/smoke/test_network_permissions.py
##########
@@ -0,0 +1,757 @@
+# 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 network permissions
+"""
+
+import logging
+
+from nose.plugins.attrib import attr
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+from marvin.lib.base import (Account,
+ Configurations,
+ Domain,
+ Project,
+ ServiceOffering,
+ VirtualMachine,
+ Zone,
+ Network,
+ NetworkOffering,
+ NetworkPermission,
+ NIC,
+ PublicIPAddress,
+ LoadBalancerRule,
+ NATRule,
+ StaticNATRule,
+
+ SSHKeyPair)
+
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template)
+
+NETWORK_FILTER_ACCOUNT = 'account'
+NETWORK_FILTER_DOMAIN = 'domain'
+NETWORK_FILTER_ACCOUNT_DOMAIN = 'accountdomain'
+NETWORK_FILTER_SHARED = 'shared'
+NETWORK_FILTER_ALL = 'all'
+
+class TestNetworkPermissions(cloudstackTestCase):
+ """
+ Test user-shared networks
+ """
+ @classmethod
+ def setUpClass(cls):
+ cls.testClient = super(
+ TestNetworkPermissions,
+ 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("TestNetworkPermissions")
+ 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["isolated_network_offering"]
+ )
+ cls.network_offering_isolated.update(cls.apiclient, state='Enabled')
+ cls._cleanup.append(cls.network_offering_isolated)
+
+ # 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.network_owner = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11A"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.network_owner)
+
+ cls.other_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11B"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.other_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.domainadmin_apiclient = cls.testClient.getUserApiClient(
+ cls.domainadmin_user.username, cls.sub_domain.name
+ )
+ cls.networkowner_user = cls.network_owner.user[0]
+ cls.user_apiclient = cls.testClient.getUserApiClient(
+ cls.networkowner_user.username, cls.sub_domain.name
+ )
+
+ cls.otheruser_user = cls.other_user.user[0]
+ cls.otheruser_apiclient = cls.testClient.getUserApiClient(
+ cls.otheruser_user.username, cls.sub_domain.name
+ )
+
+ # Create networks for domain admin, normal user and project
+ cls.services["network"]["name"] = "Test Network Isolated - Project"
+ cls.project_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ projectid=cls.project.id,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Domain
admin"
+ cls.domainadmin_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.domain_admin.name,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Normal user"
+ cls.user_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.network_owner.name,
+ zoneid=cls.zone.id
+ )
+
+ @classmethod
+ def tearDownClass(cls):
+ super().tearDownClass()
+
+ def setUp(self):
+ self.cleanup = []
+ self.virtual_machine = None
+
+ def tearDown(self):
+ super().tearDown()
Review comment:
```suggestion
super(TestNetworkPermissions, self).tearDown()
```
##########
File path: server/src/main/java/com/cloud/api/ApiResponseHelper.java
##########
@@ -3188,6 +3206,16 @@ public PrivateGatewayResponse
createPrivateGatewayResponse(PrivateGateway result
response.setAclName(acl.getName());
}
+ final NetworkDetailVO detail =
networkDetailsDao.findDetail(result.getNetworkId(),
Network.AssociatedNetworkId);
+ if (detail != null) {
+ Long associatedNetworkId = Long.valueOf(detail.getValue());
+ NetworkVO associatedNetwork =
ApiDBUtils.findNetworkById(associatedNetworkId);
+ if (associatedNetwork != null) {
+ response.setAssociatedNetworkId(associatedNetwork.getUuid());
+ response.setAssociatedNetworkName(associatedNetwork.getName());
+ }
+ }
+
Review comment:
this seems to be the same
`addAssociatedNetworkToResponse(NetworkResponse response)` as above, let's
merge it.
##########
File path: test/integration/smoke/test_user_private_gateway.py
##########
@@ -0,0 +1,425 @@
+# 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 sub-domain
+ cls.sub_domain = Domain.create(
+ cls.apiclient,
+ cls.services["acl"]["domain1"]
+ )
+
+ # 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.normal_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD1B"],
+ domainid=cls.sub_domain.id
+ )
+ # Create project
+ cls.project = Project.create(
+ cls.apiclient,
+ cls.services["project"],
+ account=cls.domain_admin.name,
+ domainid=cls.domain_admin.domainid
+ )
+
+ # Create small service offering
+ cls.service_offering = ServiceOffering.create(
+ cls.apiclient,
+ cls.services["service_offerings"]["small"]
+ )
+
+ # 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')
+
+ # Create vpc offering
+ cls.vpc_offering = VpcOffering.create(
+ cls.apiclient,
+ cls.services["vpc_offering_multi_lb"])
+ cls.vpc_offering.update(cls.apiclient, state='Enabled')
+
+ # 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')
+
+ # 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
+ )
+
+ cls._cleanup.append(cls.service_offering)
+ cls._cleanup.append(cls.network_offering_isolated)
+ cls._cleanup.append(cls.vpc_offering)
+ cls._cleanup.append(cls.network_offering_vpc)
+
+ cls._cleanup.append(cls.sub_domain)
+ cls._cleanup.append(cls.normal_user)
+ cls._cleanup.append(cls.domain_admin)
+ cls._cleanup.append(cls.project)
Review comment:
please see my comments on test_user_shared_network.py
##########
File path: test/integration/smoke/test_network_permissions.py
##########
@@ -0,0 +1,757 @@
+# 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 network permissions
+"""
+
+import logging
+
+from nose.plugins.attrib import attr
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+from marvin.lib.base import (Account,
+ Configurations,
+ Domain,
+ Project,
+ ServiceOffering,
+ VirtualMachine,
+ Zone,
+ Network,
+ NetworkOffering,
+ NetworkPermission,
+ NIC,
+ PublicIPAddress,
+ LoadBalancerRule,
+ NATRule,
+ StaticNATRule,
+
+ SSHKeyPair)
+
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template)
+
+NETWORK_FILTER_ACCOUNT = 'account'
+NETWORK_FILTER_DOMAIN = 'domain'
+NETWORK_FILTER_ACCOUNT_DOMAIN = 'accountdomain'
+NETWORK_FILTER_SHARED = 'shared'
+NETWORK_FILTER_ALL = 'all'
+
+class TestNetworkPermissions(cloudstackTestCase):
+ """
+ Test user-shared networks
+ """
+ @classmethod
+ def setUpClass(cls):
+ cls.testClient = super(
+ TestNetworkPermissions,
+ 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("TestNetworkPermissions")
+ 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["isolated_network_offering"]
+ )
+ cls.network_offering_isolated.update(cls.apiclient, state='Enabled')
+ cls._cleanup.append(cls.network_offering_isolated)
+
+ # 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.network_owner = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11A"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.network_owner)
+
+ cls.other_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11B"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.other_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.domainadmin_apiclient = cls.testClient.getUserApiClient(
+ cls.domainadmin_user.username, cls.sub_domain.name
+ )
+ cls.networkowner_user = cls.network_owner.user[0]
+ cls.user_apiclient = cls.testClient.getUserApiClient(
+ cls.networkowner_user.username, cls.sub_domain.name
+ )
+
+ cls.otheruser_user = cls.other_user.user[0]
+ cls.otheruser_apiclient = cls.testClient.getUserApiClient(
+ cls.otheruser_user.username, cls.sub_domain.name
+ )
+
+ # Create networks for domain admin, normal user and project
+ cls.services["network"]["name"] = "Test Network Isolated - Project"
+ cls.project_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ projectid=cls.project.id,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Domain
admin"
+ cls.domainadmin_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.domain_admin.name,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Normal user"
+ cls.user_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.network_owner.name,
+ zoneid=cls.zone.id
+ )
+
+ @classmethod
+ def tearDownClass(cls):
+ super().tearDownClass()
+
+ def setUp(self):
+ self.cleanup = []
+ self.virtual_machine = None
+
+ def tearDown(self):
+ super().tearDown()
+
+ def list_network(self, apiclient, account, network, project,
network_filter=None, expected=True):
+ # List networks by apiclient, account, network, project and network
network_filter
+ # If account is specified, list the networks which can be used by the
domain (canusefordeploy=true,listall=false)
+ # otherwise canusefordeploy is None and listall is True.
+ domain_id = None
+ account_name = None
+ project_id = None
+ canusefordeploy = None
+ list_all = True
+ if account:
+ domain_id = account.domainid
+ account_name = account.name
+ canusefordeploy = True
+ list_all = False
+ if project:
+ project_id = project.id
+ networks = None
+ try:
+ networks = Network.list(
+ apiclient,
+ canusefordeploy=canusefordeploy,
+ listall=list_all,
+ networkfilter= network_filter,
+ domainid=domain_id,
+ account=account_name,
+ projectid=project_id,
+ id=network.id
+ )
+ if isinstance(networks, list) and len(networks) > 0:
+ if not expected:
+ self.fail("Found the network, but expected to fail")
+ elif expected:
+ self.fail("Failed to find the network, but expected to
succeed")
+ except Exception as ex:
+ networks = None
+ if expected:
+ self.fail(f"Failed to list network, but expected to succeed :
{ex}")
+ if networks and not expected:
+ self.fail("network is listed successfully, but expected to fail")
+
+ def list_network_by_filters(self, apiclient, account, network, project,
expected_results=None):
+ # expected results in order: account/domain/accountdomain/shared/all
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ACCOUNT, expected_results[0])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_DOMAIN, expected_results[1])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ACCOUNT_DOMAIN, expected_results[2])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_SHARED, expected_results[3])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ALL, expected_results[4])
+
+ def create_network_permission(self, apiclient, network, account, project,
expected=True):
+ account_id = None
+ project_id = None
+ if account:
+ account_id = account.id
+ if project:
+ project_id = project.id
+ result = True
+ try:
+ NetworkPermission.create(
+ apiclient,
+ networkid=network.id,
+ accountids=account_id,
+ projectids=project_id
+ )
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to create network permissions, but expected
to succeed : {ex}")
+ if result and not expected:
+ self.fail("network permission is created successfully, but
expected to fail")
+
+ def remove_network_permission(self, apiclient, network, account, project,
expected=True):
+ account_id = None
+ project_id = None
+ if account:
+ account_id = account.id
+ if project:
+ project_id = project.id
+ result = True
+ try:
+ NetworkPermission.remove(
+ apiclient,
+ networkid=network.id,
+ accountids=account_id,
+ projectids=project_id
+ )
Review comment:
and removed from `self.cleanup` again?
##########
File path:
api/src/main/java/org/apache/cloudstack/api/command/user/network/CreateNetworkCmd.java
##########
@@ -249,11 +260,31 @@ public Long getPhysicalNetworkId() {
throw new InvalidParameterValueException("Unable to find network
offering by ID " + networkOfferingId);
}
+ Network associatedNetwork = null;
+ if (associatedNetworkId != null) {
+ associatedNetwork = _entityMgr.findById(Network.class,
associatedNetworkId);
+ if (associatedNetwork == null) {
+ throw new InvalidParameterValueException("Unable to find
network by ID " + associatedNetworkId);
+ }
+ if (offering.getGuestType() != GuestType.Shared) {
+ throw new InvalidParameterValueException("Associated network
ID can be specified for networks of guest IP type " + GuestType.Shared + "
only.");
+ }
+ if (zoneId != null && associatedNetwork.getDataCenterId() !=
zoneId) {
+ throw new InvalidParameterValueException("The network can only
be created in the same zone as the associated network");
+ } else if (zoneId == null) {
+ zoneId = associatedNetwork.getDataCenterId();
+ }
+ if (physicalNetworkId != null &&
!physicalNetworkId.equals(associatedNetwork.getPhysicalNetworkId())) {
+ throw new InvalidParameterValueException("The network can only
be created on the same physical network as the associated network");
+ } else if (physicalNetworkId == null) {
+ physicalNetworkId = associatedNetwork.getPhysicalNetworkId();
+ }
+ }
Review comment:
can you extract something like `getAssociationData()`? or maybe chop up
in even smaller `get..()`, `check..()` and `verify..()`parts?
##########
File path: test/integration/smoke/test_user_shared_network.py
##########
@@ -0,0 +1,630 @@
+# 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 sub-domain
+ cls.sub_domain = Domain.create(
+ cls.apiclient,
+ cls.services["acl"]["domain1"]
+ )
+
+ # 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.normal_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD1B"],
+ domainid=cls.sub_domain.id
+ )
+ # Create project
+ cls.project = Project.create(
+ cls.apiclient,
+ cls.services["project"],
+ account=cls.domain_admin.name,
+ domainid=cls.domain_admin.domainid
+ )
+
+ # Create small service offering
+ cls.service_offering = ServiceOffering.create(
+ cls.apiclient,
+ cls.services["service_offerings"]["small"]
+ )
+
+ # 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')
+
+ # 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')
+
+ # 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')
+
+ # Create vpc offering
+ cls.vpc_offering = VpcOffering.create(
+ cls.apiclient,
+ cls.services["vpc_offering_multi_lb"])
+ cls.vpc_offering.update(cls.apiclient, state='Enabled')
+
+ # 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')
+
+ # 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
+ )
+
+ cls._cleanup.append(cls.service_offering)
+ cls._cleanup.append(cls.network_offering_withvlan)
+ cls._cleanup.append(cls.network_offering_novlan)
+ cls._cleanup.append(cls.network_offering_isolated)
+ cls._cleanup.append(cls.network_offering_vpc)
+ cls._cleanup.append(cls.vpc_offering)
+
+ cls._cleanup.append(cls.sub_domain)
+ cls._cleanup.append(cls.normal_user)
+ cls._cleanup.append(cls.domain_admin)
+ cls._cleanup.append(cls.project)
+
+ @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
+ )
Review comment:
here is where I'd add the `network` to `self.cleanup`
##########
File path: test/integration/smoke/test_network_permissions.py
##########
@@ -0,0 +1,757 @@
+# 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 network permissions
+"""
+
+import logging
+
+from nose.plugins.attrib import attr
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+from marvin.lib.base import (Account,
+ Configurations,
+ Domain,
+ Project,
+ ServiceOffering,
+ VirtualMachine,
+ Zone,
+ Network,
+ NetworkOffering,
+ NetworkPermission,
+ NIC,
+ PublicIPAddress,
+ LoadBalancerRule,
+ NATRule,
+ StaticNATRule,
+
+ SSHKeyPair)
+
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template)
+
+NETWORK_FILTER_ACCOUNT = 'account'
+NETWORK_FILTER_DOMAIN = 'domain'
+NETWORK_FILTER_ACCOUNT_DOMAIN = 'accountdomain'
+NETWORK_FILTER_SHARED = 'shared'
+NETWORK_FILTER_ALL = 'all'
+
+class TestNetworkPermissions(cloudstackTestCase):
+ """
+ Test user-shared networks
+ """
+ @classmethod
+ def setUpClass(cls):
+ cls.testClient = super(
+ TestNetworkPermissions,
+ 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("TestNetworkPermissions")
+ 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["isolated_network_offering"]
+ )
+ cls.network_offering_isolated.update(cls.apiclient, state='Enabled')
+ cls._cleanup.append(cls.network_offering_isolated)
+
+ # 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.network_owner = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11A"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.network_owner)
+
+ cls.other_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11B"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.other_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.domainadmin_apiclient = cls.testClient.getUserApiClient(
+ cls.domainadmin_user.username, cls.sub_domain.name
+ )
+ cls.networkowner_user = cls.network_owner.user[0]
+ cls.user_apiclient = cls.testClient.getUserApiClient(
+ cls.networkowner_user.username, cls.sub_domain.name
+ )
+
+ cls.otheruser_user = cls.other_user.user[0]
+ cls.otheruser_apiclient = cls.testClient.getUserApiClient(
+ cls.otheruser_user.username, cls.sub_domain.name
+ )
+
+ # Create networks for domain admin, normal user and project
+ cls.services["network"]["name"] = "Test Network Isolated - Project"
+ cls.project_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ projectid=cls.project.id,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Domain
admin"
+ cls.domainadmin_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.domain_admin.name,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Normal user"
+ cls.user_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.network_owner.name,
+ zoneid=cls.zone.id
+ )
+
+ @classmethod
+ def tearDownClass(cls):
+ super().tearDownClass()
+
+ def setUp(self):
+ self.cleanup = []
+ self.virtual_machine = None
+
+ def tearDown(self):
+ super().tearDown()
+
+ def list_network(self, apiclient, account, network, project,
network_filter=None, expected=True):
+ # List networks by apiclient, account, network, project and network
network_filter
+ # If account is specified, list the networks which can be used by the
domain (canusefordeploy=true,listall=false)
+ # otherwise canusefordeploy is None and listall is True.
+ domain_id = None
+ account_name = None
+ project_id = None
+ canusefordeploy = None
+ list_all = True
+ if account:
+ domain_id = account.domainid
+ account_name = account.name
+ canusefordeploy = True
+ list_all = False
+ if project:
+ project_id = project.id
+ networks = None
+ try:
+ networks = Network.list(
+ apiclient,
+ canusefordeploy=canusefordeploy,
+ listall=list_all,
+ networkfilter= network_filter,
+ domainid=domain_id,
+ account=account_name,
+ projectid=project_id,
+ id=network.id
+ )
+ if isinstance(networks, list) and len(networks) > 0:
+ if not expected:
+ self.fail("Found the network, but expected to fail")
+ elif expected:
+ self.fail("Failed to find the network, but expected to
succeed")
+ except Exception as ex:
+ networks = None
+ if expected:
+ self.fail(f"Failed to list network, but expected to succeed :
{ex}")
+ if networks and not expected:
+ self.fail("network is listed successfully, but expected to fail")
+
+ def list_network_by_filters(self, apiclient, account, network, project,
expected_results=None):
+ # expected results in order: account/domain/accountdomain/shared/all
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ACCOUNT, expected_results[0])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_DOMAIN, expected_results[1])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ACCOUNT_DOMAIN, expected_results[2])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_SHARED, expected_results[3])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ALL, expected_results[4])
+
+ def create_network_permission(self, apiclient, network, account, project,
expected=True):
+ account_id = None
+ project_id = None
+ if account:
+ account_id = account.id
+ if project:
+ project_id = project.id
+ result = True
+ try:
+ NetworkPermission.create(
+ apiclient,
+ networkid=network.id,
+ accountids=account_id,
+ projectids=project_id
+ )
Review comment:
should this be cleaned (added to `self.cleanup`)
##########
File path: test/integration/smoke/test_network_permissions.py
##########
@@ -0,0 +1,757 @@
+# 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 network permissions
+"""
+
+import logging
+
+from nose.plugins.attrib import attr
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+from marvin.lib.base import (Account,
+ Configurations,
+ Domain,
+ Project,
+ ServiceOffering,
+ VirtualMachine,
+ Zone,
+ Network,
+ NetworkOffering,
+ NetworkPermission,
+ NIC,
+ PublicIPAddress,
+ LoadBalancerRule,
+ NATRule,
+ StaticNATRule,
+
+ SSHKeyPair)
+
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template)
+
+NETWORK_FILTER_ACCOUNT = 'account'
+NETWORK_FILTER_DOMAIN = 'domain'
+NETWORK_FILTER_ACCOUNT_DOMAIN = 'accountdomain'
+NETWORK_FILTER_SHARED = 'shared'
+NETWORK_FILTER_ALL = 'all'
+
+class TestNetworkPermissions(cloudstackTestCase):
+ """
+ Test user-shared networks
+ """
+ @classmethod
+ def setUpClass(cls):
+ cls.testClient = super(
+ TestNetworkPermissions,
+ 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("TestNetworkPermissions")
+ 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["isolated_network_offering"]
+ )
+ cls.network_offering_isolated.update(cls.apiclient, state='Enabled')
+ cls._cleanup.append(cls.network_offering_isolated)
+
+ # 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.network_owner = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11A"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.network_owner)
+
+ cls.other_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11B"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.other_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.domainadmin_apiclient = cls.testClient.getUserApiClient(
+ cls.domainadmin_user.username, cls.sub_domain.name
+ )
+ cls.networkowner_user = cls.network_owner.user[0]
+ cls.user_apiclient = cls.testClient.getUserApiClient(
+ cls.networkowner_user.username, cls.sub_domain.name
+ )
+
+ cls.otheruser_user = cls.other_user.user[0]
+ cls.otheruser_apiclient = cls.testClient.getUserApiClient(
+ cls.otheruser_user.username, cls.sub_domain.name
+ )
+
+ # Create networks for domain admin, normal user and project
+ cls.services["network"]["name"] = "Test Network Isolated - Project"
+ cls.project_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ projectid=cls.project.id,
+ zoneid=cls.zone.id
+ )
Review comment:
should this be cleaned (added to `cls._cleanup`)
##########
File path: test/integration/smoke/test_network_permissions.py
##########
@@ -0,0 +1,757 @@
+# 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 network permissions
+"""
+
+import logging
+
+from nose.plugins.attrib import attr
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+from marvin.lib.base import (Account,
+ Configurations,
+ Domain,
+ Project,
+ ServiceOffering,
+ VirtualMachine,
+ Zone,
+ Network,
+ NetworkOffering,
+ NetworkPermission,
+ NIC,
+ PublicIPAddress,
+ LoadBalancerRule,
+ NATRule,
+ StaticNATRule,
+
+ SSHKeyPair)
+
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template)
+
+NETWORK_FILTER_ACCOUNT = 'account'
+NETWORK_FILTER_DOMAIN = 'domain'
+NETWORK_FILTER_ACCOUNT_DOMAIN = 'accountdomain'
+NETWORK_FILTER_SHARED = 'shared'
+NETWORK_FILTER_ALL = 'all'
+
+class TestNetworkPermissions(cloudstackTestCase):
+ """
+ Test user-shared networks
+ """
+ @classmethod
+ def setUpClass(cls):
+ cls.testClient = super(
+ TestNetworkPermissions,
+ 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("TestNetworkPermissions")
+ 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["isolated_network_offering"]
+ )
+ cls.network_offering_isolated.update(cls.apiclient, state='Enabled')
+ cls._cleanup.append(cls.network_offering_isolated)
+
+ # 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.network_owner = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11A"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.network_owner)
+
+ cls.other_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11B"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.other_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.domainadmin_apiclient = cls.testClient.getUserApiClient(
+ cls.domainadmin_user.username, cls.sub_domain.name
+ )
+ cls.networkowner_user = cls.network_owner.user[0]
+ cls.user_apiclient = cls.testClient.getUserApiClient(
+ cls.networkowner_user.username, cls.sub_domain.name
+ )
+
+ cls.otheruser_user = cls.other_user.user[0]
+ cls.otheruser_apiclient = cls.testClient.getUserApiClient(
+ cls.otheruser_user.username, cls.sub_domain.name
+ )
+
+ # Create networks for domain admin, normal user and project
+ cls.services["network"]["name"] = "Test Network Isolated - Project"
+ cls.project_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ projectid=cls.project.id,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Domain
admin"
+ cls.domainadmin_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.domain_admin.name,
+ zoneid=cls.zone.id
+ )
Review comment:
should this be cleaned (added to `cls._cleanup`)
##########
File path: server/src/main/java/com/cloud/network/vpc/VpcManagerImpl.java
##########
@@ -1833,8 +1860,63 @@ public PrivateGateway createVpcPrivateGateway(final long
vpcId, Long physicalNet
throw ex;
}
+ // Validate vlanId and associatedNetworkId
+ if (broadcastUri == null && associatedNetworkId == null) {
+ throw new InvalidParameterValueException("One of vlanId and
associatedNetworkId must be specified");
+ }
+ if (broadcastUri != null && associatedNetworkId != null) {
+ throw new InvalidParameterValueException("vlanId and
associatedNetworkId are mutually exclusive");
+ }
+
+ // Validate network offering
+ NetworkOfferingVO ntwkOff = null;
+ if (networkOfferingIdPassed != null) {
+ ntwkOff = _networkOfferingDao.findById(networkOfferingIdPassed);
+ if (ntwkOff == null) {
+ throw new InvalidParameterValueException("Unable to find
network offering by id specified");
+ }
+ if (! TrafficType.Guest.equals(ntwkOff.getTrafficType())) {
+ throw new InvalidParameterValueException("The network offering
cannot be used to create Guest network");
+ }
+ if (! GuestType.Isolated.equals(ntwkOff.getGuestType())) {
+ throw new InvalidParameterValueException("The network offering
cannot be used to create Isolated network");
+ }
+ } else if (broadcastUri != null) {
+ ntwkOff =
_networkOfferingDao.findByUniqueName(NetworkOffering.SystemPrivateGatewayNetworkOffering);
+ } else {
+ ntwkOff =
_networkOfferingDao.findByUniqueName(NetworkOffering.SystemPrivateGatewayNetworkOfferingWithoutVlan);
+ }
+ final Long networkOfferingId = ntwkOff.getId();
+
+ Account caller = CallContext.current().getCallingAccount();
+ if (!_accountMgr.isRootAdmin(caller.getId()) &&
(ntwkOff.isSpecifyVlan() || broadcastUri != null || bypassVlanOverlapCheck)) {
+ throw new InvalidParameterValueException("Only ROOT admin is
allowed to specify vlanId or bypass vlan overlap check");
+ }
+ if (ntwkOff.isSpecifyVlan() && broadcastUri == null) {
+ throw new InvalidParameterValueException("vlanId must be specified
for this network offering");
+ }
+ if (! ntwkOff.isSpecifyVlan() && associatedNetworkId == null) {
+ throw new InvalidParameterValueException("associatedNetworkId must
be specified for this network offering");
+ }
+
+ final Long dcId = vpc.getZoneId();
PhysicalNetwork physNet = null;
// Validate physical network
+ if (associatedNetworkId != null) {
+ Network associatedNetwork = _entityMgr.findById(Network.class,
associatedNetworkId);
+ if (associatedNetwork == null) {
+ throw new InvalidParameterValueException("Unable to find
network by ID " + associatedNetworkId);
+ }
+ if (physicalNetworkId != null &&
!physicalNetworkId.equals(associatedNetwork.getPhysicalNetworkId())) {
+ throw new InvalidParameterValueException("The network can only
be created on the same physical network as the associated network");
+ } else if (physicalNetworkId == null) {
+ physicalNetworkId = associatedNetwork.getPhysicalNetworkId();
+ }
+ }
+ if (physicalNetworkId == null) {
+ // Determine the physical network by network offering tags
+ physicalNetworkId = _ntwkSvc.findPhysicalNetworkId(dcId,
ntwkOff.getTags(), ntwkOff.getTrafficType());
+ }
Review comment:
can you create a new method for this?
##########
File path: server/src/main/java/com/cloud/network/vpc/VpcManagerImpl.java
##########
@@ -1833,8 +1860,63 @@ public PrivateGateway createVpcPrivateGateway(final long
vpcId, Long physicalNet
throw ex;
}
+ // Validate vlanId and associatedNetworkId
+ if (broadcastUri == null && associatedNetworkId == null) {
+ throw new InvalidParameterValueException("One of vlanId and
associatedNetworkId must be specified");
+ }
+ if (broadcastUri != null && associatedNetworkId != null) {
+ throw new InvalidParameterValueException("vlanId and
associatedNetworkId are mutually exclusive");
+ }
Review comment:
extra `verify...()` to put this in
##########
File path: test/integration/smoke/test_network_permissions.py
##########
@@ -0,0 +1,757 @@
+# 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 network permissions
+"""
+
+import logging
+
+from nose.plugins.attrib import attr
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+from marvin.lib.base import (Account,
+ Configurations,
+ Domain,
+ Project,
+ ServiceOffering,
+ VirtualMachine,
+ Zone,
+ Network,
+ NetworkOffering,
+ NetworkPermission,
+ NIC,
+ PublicIPAddress,
+ LoadBalancerRule,
+ NATRule,
+ StaticNATRule,
+
+ SSHKeyPair)
+
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template)
+
+NETWORK_FILTER_ACCOUNT = 'account'
+NETWORK_FILTER_DOMAIN = 'domain'
+NETWORK_FILTER_ACCOUNT_DOMAIN = 'accountdomain'
+NETWORK_FILTER_SHARED = 'shared'
+NETWORK_FILTER_ALL = 'all'
+
+class TestNetworkPermissions(cloudstackTestCase):
+ """
+ Test user-shared networks
+ """
+ @classmethod
+ def setUpClass(cls):
+ cls.testClient = super(
+ TestNetworkPermissions,
+ 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("TestNetworkPermissions")
+ 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["isolated_network_offering"]
+ )
+ cls.network_offering_isolated.update(cls.apiclient, state='Enabled')
+ cls._cleanup.append(cls.network_offering_isolated)
+
+ # 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.network_owner = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11A"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.network_owner)
+
+ cls.other_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11B"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.other_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.domainadmin_apiclient = cls.testClient.getUserApiClient(
+ cls.domainadmin_user.username, cls.sub_domain.name
+ )
+ cls.networkowner_user = cls.network_owner.user[0]
+ cls.user_apiclient = cls.testClient.getUserApiClient(
+ cls.networkowner_user.username, cls.sub_domain.name
+ )
+
+ cls.otheruser_user = cls.other_user.user[0]
+ cls.otheruser_apiclient = cls.testClient.getUserApiClient(
+ cls.otheruser_user.username, cls.sub_domain.name
+ )
+
+ # Create networks for domain admin, normal user and project
+ cls.services["network"]["name"] = "Test Network Isolated - Project"
+ cls.project_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ projectid=cls.project.id,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Domain
admin"
+ cls.domainadmin_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.domain_admin.name,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Normal user"
+ cls.user_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.network_owner.name,
+ zoneid=cls.zone.id
+ )
+
+ @classmethod
+ def tearDownClass(cls):
+ super().tearDownClass()
+
+ def setUp(self):
+ self.cleanup = []
+ self.virtual_machine = None
+
+ def tearDown(self):
+ super().tearDown()
+
+ def list_network(self, apiclient, account, network, project,
network_filter=None, expected=True):
+ # List networks by apiclient, account, network, project and network
network_filter
+ # If account is specified, list the networks which can be used by the
domain (canusefordeploy=true,listall=false)
+ # otherwise canusefordeploy is None and listall is True.
+ domain_id = None
+ account_name = None
+ project_id = None
+ canusefordeploy = None
+ list_all = True
+ if account:
+ domain_id = account.domainid
+ account_name = account.name
+ canusefordeploy = True
+ list_all = False
+ if project:
+ project_id = project.id
+ networks = None
+ try:
+ networks = Network.list(
+ apiclient,
+ canusefordeploy=canusefordeploy,
+ listall=list_all,
+ networkfilter= network_filter,
+ domainid=domain_id,
+ account=account_name,
+ projectid=project_id,
+ id=network.id
+ )
+ if isinstance(networks, list) and len(networks) > 0:
+ if not expected:
+ self.fail("Found the network, but expected to fail")
+ elif expected:
+ self.fail("Failed to find the network, but expected to
succeed")
+ except Exception as ex:
+ networks = None
+ if expected:
+ self.fail(f"Failed to list network, but expected to succeed :
{ex}")
+ if networks and not expected:
+ self.fail("network is listed successfully, but expected to fail")
+
+ def list_network_by_filters(self, apiclient, account, network, project,
expected_results=None):
+ # expected results in order: account/domain/accountdomain/shared/all
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ACCOUNT, expected_results[0])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_DOMAIN, expected_results[1])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ACCOUNT_DOMAIN, expected_results[2])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_SHARED, expected_results[3])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ALL, expected_results[4])
+
+ def create_network_permission(self, apiclient, network, account, project,
expected=True):
+ account_id = None
+ project_id = None
+ if account:
+ account_id = account.id
+ if project:
+ project_id = project.id
+ result = True
+ try:
+ NetworkPermission.create(
+ apiclient,
+ networkid=network.id,
+ accountids=account_id,
+ projectids=project_id
+ )
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to create network permissions, but expected
to succeed : {ex}")
+ if result and not expected:
+ self.fail("network permission is created successfully, but
expected to fail")
+
+ def remove_network_permission(self, apiclient, network, account, project,
expected=True):
+ account_id = None
+ project_id = None
+ if account:
+ account_id = account.id
+ if project:
+ project_id = project.id
+ result = True
+ try:
+ NetworkPermission.remove(
+ apiclient,
+ networkid=network.id,
+ accountids=account_id,
+ projectids=project_id
+ )
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to remove network permissions, but expected
to succeed : {ex}")
+ if result and not expected:
+ self.fail("network permission is removed successfully, but
expected to fail")
+
+ def reset_network_permission(self, apiclient, network, expected=True):
+ result = True
+ try:
+ NetworkPermission.reset(
+ apiclient,
+ networkid=network.id
+ )
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to reset network permissions, but expected
to succeed : {ex}")
+ if result and not expected:
+ self.fail("network permission is reset successfully, but expected
to fail")
+
+ def exec_command(self, apiclient_str, command, expected=None):
+ result = True
+ try:
+ command = command.format(apiclient = apiclient_str)
+ exec(command)
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to execute command '{command}' with
exception : {ex}")
+ if result and expected is False:
+ self.fail(f"command {command} is executed successfully, but
expected to fail")
+ if expected is None:
+ # if expected is None, display the command and result
+ self.logger.info(f"Result of command '{command}' : {result}")
+ return result
Review comment:
how can we make sure anything created in these commands is cleaned up
after test execution finishes?
##########
File path: server/src/main/java/com/cloud/network/vpc/VpcManagerImpl.java
##########
@@ -1833,8 +1860,63 @@ public PrivateGateway createVpcPrivateGateway(final long
vpcId, Long physicalNet
throw ex;
}
+ // Validate vlanId and associatedNetworkId
+ if (broadcastUri == null && associatedNetworkId == null) {
+ throw new InvalidParameterValueException("One of vlanId and
associatedNetworkId must be specified");
+ }
+ if (broadcastUri != null && associatedNetworkId != null) {
+ throw new InvalidParameterValueException("vlanId and
associatedNetworkId are mutually exclusive");
+ }
+
+ // Validate network offering
+ NetworkOfferingVO ntwkOff = null;
+ if (networkOfferingIdPassed != null) {
+ ntwkOff = _networkOfferingDao.findById(networkOfferingIdPassed);
+ if (ntwkOff == null) {
+ throw new InvalidParameterValueException("Unable to find
network offering by id specified");
+ }
+ if (! TrafficType.Guest.equals(ntwkOff.getTrafficType())) {
+ throw new InvalidParameterValueException("The network offering
cannot be used to create Guest network");
+ }
+ if (! GuestType.Isolated.equals(ntwkOff.getGuestType())) {
+ throw new InvalidParameterValueException("The network offering
cannot be used to create Isolated network");
+ }
+ } else if (broadcastUri != null) {
+ ntwkOff =
_networkOfferingDao.findByUniqueName(NetworkOffering.SystemPrivateGatewayNetworkOffering);
+ } else {
+ ntwkOff =
_networkOfferingDao.findByUniqueName(NetworkOffering.SystemPrivateGatewayNetworkOfferingWithoutVlan);
+ }
+ final Long networkOfferingId = ntwkOff.getId();
+
+ Account caller = CallContext.current().getCallingAccount();
+ if (!_accountMgr.isRootAdmin(caller.getId()) &&
(ntwkOff.isSpecifyVlan() || broadcastUri != null || bypassVlanOverlapCheck)) {
+ throw new InvalidParameterValueException("Only ROOT admin is
allowed to specify vlanId or bypass vlan overlap check");
+ }
+ if (ntwkOff.isSpecifyVlan() && broadcastUri == null) {
+ throw new InvalidParameterValueException("vlanId must be specified
for this network offering");
+ }
+ if (! ntwkOff.isSpecifyVlan() && associatedNetworkId == null) {
+ throw new InvalidParameterValueException("associatedNetworkId must
be specified for this network offering");
+ }
Review comment:
extra `validation()`?
##########
File path: server/src/main/java/com/cloud/network/vpc/VpcManagerImpl.java
##########
@@ -1833,8 +1860,63 @@ public PrivateGateway createVpcPrivateGateway(final long
vpcId, Long physicalNet
throw ex;
}
+ // Validate vlanId and associatedNetworkId
+ if (broadcastUri == null && associatedNetworkId == null) {
+ throw new InvalidParameterValueException("One of vlanId and
associatedNetworkId must be specified");
+ }
+ if (broadcastUri != null && associatedNetworkId != null) {
+ throw new InvalidParameterValueException("vlanId and
associatedNetworkId are mutually exclusive");
+ }
+
+ // Validate network offering
+ NetworkOfferingVO ntwkOff = null;
+ if (networkOfferingIdPassed != null) {
+ ntwkOff = _networkOfferingDao.findById(networkOfferingIdPassed);
+ if (ntwkOff == null) {
+ throw new InvalidParameterValueException("Unable to find
network offering by id specified");
+ }
+ if (! TrafficType.Guest.equals(ntwkOff.getTrafficType())) {
+ throw new InvalidParameterValueException("The network offering
cannot be used to create Guest network");
+ }
+ if (! GuestType.Isolated.equals(ntwkOff.getGuestType())) {
+ throw new InvalidParameterValueException("The network offering
cannot be used to create Isolated network");
+ }
+ } else if (broadcastUri != null) {
+ ntwkOff =
_networkOfferingDao.findByUniqueName(NetworkOffering.SystemPrivateGatewayNetworkOffering);
+ } else {
+ ntwkOff =
_networkOfferingDao.findByUniqueName(NetworkOffering.SystemPrivateGatewayNetworkOfferingWithoutVlan);
+ }
Review comment:
extra method?
##########
File path: test/integration/smoke/test_network_permissions.py
##########
@@ -0,0 +1,757 @@
+# 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 network permissions
+"""
+
+import logging
+
+from nose.plugins.attrib import attr
+from marvin.cloudstackTestCase import cloudstackTestCase
+
+from marvin.lib.base import (Account,
+ Configurations,
+ Domain,
+ Project,
+ ServiceOffering,
+ VirtualMachine,
+ Zone,
+ Network,
+ NetworkOffering,
+ NetworkPermission,
+ NIC,
+ PublicIPAddress,
+ LoadBalancerRule,
+ NATRule,
+ StaticNATRule,
+
+ SSHKeyPair)
+
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template)
+
+NETWORK_FILTER_ACCOUNT = 'account'
+NETWORK_FILTER_DOMAIN = 'domain'
+NETWORK_FILTER_ACCOUNT_DOMAIN = 'accountdomain'
+NETWORK_FILTER_SHARED = 'shared'
+NETWORK_FILTER_ALL = 'all'
+
+class TestNetworkPermissions(cloudstackTestCase):
+ """
+ Test user-shared networks
+ """
+ @classmethod
+ def setUpClass(cls):
+ cls.testClient = super(
+ TestNetworkPermissions,
+ 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("TestNetworkPermissions")
+ 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["isolated_network_offering"]
+ )
+ cls.network_offering_isolated.update(cls.apiclient, state='Enabled')
+ cls._cleanup.append(cls.network_offering_isolated)
+
+ # 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.network_owner = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11A"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.network_owner)
+
+ cls.other_user = Account.create(
+ cls.apiclient,
+ cls.services["acl"]["accountD11B"],
+ domainid=cls.sub_domain.id
+ )
+ cls._cleanup.append(cls.other_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.domainadmin_apiclient = cls.testClient.getUserApiClient(
+ cls.domainadmin_user.username, cls.sub_domain.name
+ )
+ cls.networkowner_user = cls.network_owner.user[0]
+ cls.user_apiclient = cls.testClient.getUserApiClient(
+ cls.networkowner_user.username, cls.sub_domain.name
+ )
+
+ cls.otheruser_user = cls.other_user.user[0]
+ cls.otheruser_apiclient = cls.testClient.getUserApiClient(
+ cls.otheruser_user.username, cls.sub_domain.name
+ )
+
+ # Create networks for domain admin, normal user and project
+ cls.services["network"]["name"] = "Test Network Isolated - Project"
+ cls.project_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ projectid=cls.project.id,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Domain
admin"
+ cls.domainadmin_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.domain_admin.name,
+ zoneid=cls.zone.id
+ )
+
+ cls.services["network"]["name"] = "Test Network Isolated - Normal user"
+ cls.user_network = Network.create(
+ cls.apiclient,
+ cls.services["network"],
+ networkofferingid=cls.network_offering_isolated.id,
+ domainid=cls.sub_domain.id,
+ accountid=cls.network_owner.name,
+ zoneid=cls.zone.id
+ )
+
+ @classmethod
+ def tearDownClass(cls):
+ super().tearDownClass()
+
+ def setUp(self):
+ self.cleanup = []
+ self.virtual_machine = None
+
+ def tearDown(self):
+ super().tearDown()
+
+ def list_network(self, apiclient, account, network, project,
network_filter=None, expected=True):
+ # List networks by apiclient, account, network, project and network
network_filter
+ # If account is specified, list the networks which can be used by the
domain (canusefordeploy=true,listall=false)
+ # otherwise canusefordeploy is None and listall is True.
+ domain_id = None
+ account_name = None
+ project_id = None
+ canusefordeploy = None
+ list_all = True
+ if account:
+ domain_id = account.domainid
+ account_name = account.name
+ canusefordeploy = True
+ list_all = False
+ if project:
+ project_id = project.id
+ networks = None
+ try:
+ networks = Network.list(
+ apiclient,
+ canusefordeploy=canusefordeploy,
+ listall=list_all,
+ networkfilter= network_filter,
+ domainid=domain_id,
+ account=account_name,
+ projectid=project_id,
+ id=network.id
+ )
+ if isinstance(networks, list) and len(networks) > 0:
+ if not expected:
+ self.fail("Found the network, but expected to fail")
+ elif expected:
+ self.fail("Failed to find the network, but expected to
succeed")
+ except Exception as ex:
+ networks = None
+ if expected:
+ self.fail(f"Failed to list network, but expected to succeed :
{ex}")
+ if networks and not expected:
+ self.fail("network is listed successfully, but expected to fail")
+
+ def list_network_by_filters(self, apiclient, account, network, project,
expected_results=None):
+ # expected results in order: account/domain/accountdomain/shared/all
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ACCOUNT, expected_results[0])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_DOMAIN, expected_results[1])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ACCOUNT_DOMAIN, expected_results[2])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_SHARED, expected_results[3])
+ self.list_network(apiclient, account, network, project,
NETWORK_FILTER_ALL, expected_results[4])
+
+ def create_network_permission(self, apiclient, network, account, project,
expected=True):
+ account_id = None
+ project_id = None
+ if account:
+ account_id = account.id
+ if project:
+ project_id = project.id
+ result = True
+ try:
+ NetworkPermission.create(
+ apiclient,
+ networkid=network.id,
+ accountids=account_id,
+ projectids=project_id
+ )
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to create network permissions, but expected
to succeed : {ex}")
+ if result and not expected:
+ self.fail("network permission is created successfully, but
expected to fail")
+
+ def remove_network_permission(self, apiclient, network, account, project,
expected=True):
+ account_id = None
+ project_id = None
+ if account:
+ account_id = account.id
+ if project:
+ project_id = project.id
+ result = True
+ try:
+ NetworkPermission.remove(
+ apiclient,
+ networkid=network.id,
+ accountids=account_id,
+ projectids=project_id
+ )
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to remove network permissions, but expected
to succeed : {ex}")
+ if result and not expected:
+ self.fail("network permission is removed successfully, but
expected to fail")
+
+ def reset_network_permission(self, apiclient, network, expected=True):
+ result = True
+ try:
+ NetworkPermission.reset(
+ apiclient,
+ networkid=network.id
+ )
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to reset network permissions, but expected
to succeed : {ex}")
+ if result and not expected:
+ self.fail("network permission is reset successfully, but expected
to fail")
+
+ def exec_command(self, apiclient_str, command, expected=None):
+ result = True
+ try:
+ command = command.format(apiclient = apiclient_str)
+ exec(command)
+ except Exception as ex:
+ result = False
+ if expected:
+ self.fail(f"Failed to execute command '{command}' with
exception : {ex}")
+ if result and expected is False:
+ self.fail(f"command {command} is executed successfully, but
expected to fail")
+ if expected is None:
+ # if expected is None, display the command and result
+ self.logger.info(f"Result of command '{command}' : {result}")
+ return result
+
+ @attr(tags=["advanced"], required_hardware="false")
+ def test_01_network_permission_on_project_network(self):
+ """ Testing network permissions on project network """
+
+ self.create_network_permission(self.apiclient, self.project_network,
self.domain_admin, None, expected=False)
+ self.create_network_permission(self.domainadmin_apiclient,
self.project_network, self.domain_admin, None, expected=False)
+ self.create_network_permission(self.user_apiclient,
self.project_network, self.network_owner, None, expected=False)
+
+ @attr(tags=["advanced"], required_hardware="false")
+ def test_02_network_permission_on_user_network(self):
+ """ Testing network permissions on user network """
+
+ # List user network by domain admin
+ self.list_network_by_filters(self.domainadmin_apiclient, None,
self.user_network, None, [True, False, True, False, True])
+ self.list_network_by_filters(self.domainadmin_apiclient,
self.domain_admin, self.user_network, None, [False, False, False, False, False])
+
+ # Create network permissions
+ self.create_network_permission(self.apiclient, self.user_network,
self.domain_admin, None, expected=True)
+ self.create_network_permission(self.domainadmin_apiclient,
self.user_network, self.domain_admin, None, expected=True)
+ self.create_network_permission(self.user_apiclient, self.user_network,
self.network_owner, None, expected=True)
+ self.create_network_permission(self.user_apiclient, self.user_network,
self.other_user, None, expected=True)
+ self.create_network_permission(self.user_apiclient, self.user_network,
None, self.project, expected=False)
+ self.create_network_permission(self.domainadmin_apiclient,
self.user_network, None, self.project, expected=True)
+ self.create_network_permission(self.otheruser_apiclient,
self.user_network, self.network_owner, None, expected=False)
+
+ # List domain admin network by domain admin
+ self.list_network_by_filters(self.domainadmin_apiclient, None,
self.domainadmin_network, None, [True, False, True, False, True])
+ self.list_network_by_filters(self.domainadmin_apiclient,
self.domain_admin, self.domainadmin_network, None, [True, False, True, False,
True])
+ # List user network by domain admin
+ self.list_network_by_filters(self.domainadmin_apiclient, None,
self.user_network, None, [True, False, True, True, True])
+ self.list_network_by_filters(self.domainadmin_apiclient,
self.domain_admin, self.user_network, None, [False, False, False, True, True])
+ # List user network by user
+ self.list_network_by_filters(self.user_apiclient, None,
self.user_network, None, [True, False, True, False, True])
+ self.list_network_by_filters(self.user_apiclient, self.network_owner,
self.user_network, None, [True, False, True, False, True])
+ # List user network by other user
+ self.list_network_by_filters(self.otheruser_apiclient, None,
self.user_network, None, [False, False, False, True, True])
+ self.list_network_by_filters(self.otheruser_apiclient,
self.network_owner, self.user_network, None, [False, False, False, False,
False])
+
+ # Remove network permissions
+ self.remove_network_permission(self.domainadmin_apiclient,
self.user_network, self.domain_admin, None, expected=True)
+ # List user network by domain admin
+ self.list_network_by_filters(self.domainadmin_apiclient, None,
self.user_network, None, [True, False, True, True, True])
+ self.list_network_by_filters(self.domainadmin_apiclient,
self.domain_admin, self.user_network, None, [False, False, False, False, False])
+
+ # Reset network permissions
+ self.reset_network_permission(self.domainadmin_apiclient,
self.user_network, expected=True)
+ # List user network by domain admin
+ self.list_network_by_filters(self.domainadmin_apiclient, None,
self.user_network, None, [True, False, True, False, True])
+ self.list_network_by_filters(self.domainadmin_apiclient,
self.domain_admin, self.user_network, None, [False, False, False, False, False])
+
+ @attr(tags=["advanced"], required_hardware="false")
+ def test_03_network_operations_on_created_vm_of_otheruser(self):
+ """ Testing network operations on a create vm owned by other user"""
+
+ # 1. Create an Isolated network by other user
+ self.services["network"]["name"] = "Test Network Isolated - Other user"
+ otheruser_network = Network.create(
+ self.otheruser_apiclient,
+ self.services["network"],
+ networkofferingid=self.network_offering_isolated.id,
+ zoneid=self.zone.id
+ )
+ self.cleanup = [otheruser_network]
Review comment:
```suggestion
self.cleanup.append(otheruser_network)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
