You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by gi...@apache.org on 2013/12/27 11:31:57 UTC
[1/3] CLOUDSTACK-2237: Automation - Adding new test cases for
security group in advanced zone feature
Updated Branches:
refs/heads/4.3 f908a1575 -> 029f4e36a
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/029f4e36/tools/marvin/marvin/config/config.cfg
----------------------------------------------------------------------
diff --git a/tools/marvin/marvin/config/config.cfg b/tools/marvin/marvin/config/config.cfg
new file mode 100644
index 0000000..5849fe8
--- /dev/null
+++ b/tools/marvin/marvin/config/config.cfg
@@ -0,0 +1,159 @@
+# 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.
+#
+# Use the common configs added such as account, network_offerings, domain, project,
+# or add your own data if required separately for any test case
+{
+ "domain":
+ {
+ "name": "domain"
+ }
+ ,
+ "project":
+ {
+ "name": "Project",
+ "displaytext": "Test project"
+ },
+ "account": {
+ "email": "test-account@test.com",
+ "firstname": "test",
+ "lastname": "test",
+ "username": "test-account",
+ "password": "password"
+ },
+ "service_offering": {
+ "name": "Tiny Instance",
+ "displaytext": "Tiny Instance",
+ "cpunumber": 1,
+ "cpuspeed": 100,
+ "memory": 128
+ },
+ "isolated_network_offering": {
+ "name": "Network offering-DA services",
+ "displaytext": "Network offering-DA services",
+ "guestiptype": "Isolated",
+ "supportedservices": "Dhcp,Dns,SourceNat,PortForwarding,Vpn,Firewall,Lb,UserData,StaticNat",
+ "traffictype": "GUEST",
+ "availability": "Optional'",
+ "serviceProviderList": {
+ "Dhcp": "VirtualRouter",
+ "Dns": "VirtualRouter",
+ "SourceNat": "VirtualRouter",
+ "PortForwarding": "VirtualRouter",
+ "Vpn": "VirtualRouter",
+ "Firewall": "VirtualRouter",
+ "Lb": "VirtualRouter",
+ "UserData": "VirtualRouter",
+ "StaticNat": "VirtualRouter"
+ }
+ },
+ "isolated_network": {
+ "name": "Isolated Network",
+ "displaytext": "Isolated Network"
+ },
+ "virtual_machine": {
+ "displayname": "Test VM",
+ "username": "root",
+ "password": "password",
+ "ssh_port": 22,
+ "privateport": 22,
+ "publicport": 22,
+ "protocol": "TCP"
+ },
+ "shared_network": {
+ "name": "MySharedNetwork - Test",
+ "displaytext": "MySharedNetwork",
+ "vlan" : "",
+ "gateway" :"",
+ "netmask" :"",
+ "startip" :"",
+ "endip" :"",
+ "acltype" : "Domain",
+ "scope":"all"
+ },
+ "shared_network_offering_sg": {
+ "name": "MySharedOffering-sg",
+ "displaytext": "MySharedOffering-sg",
+ "guestiptype": "Shared",
+ "supportedservices": "Dhcp,Dns,UserData,SecurityGroup",
+ "specifyVlan" : "False",
+ "specifyIpRanges" : "False",
+ "traffictype": "GUEST",
+ "serviceProviderList" : {
+ "Dhcp": "VirtualRouter",
+ "Dns": "VirtualRouter",
+ "UserData": "VirtualRouter",
+ "SecurityGroup": "SecurityGroupProvider"
+ }
+ },
+ "shared_network_sg": {
+ "name": "Shared-Network-SG-Test",
+ "displaytext": "Shared-Network_SG-Test",
+ "networkofferingid":"1",
+ "vlan" : "",
+ "gateway" :"",
+ "netmask" :"255.255.255.0",
+ "startip" :"",
+ "endip" :"",
+ "acltype" : "Domain",
+ "scope":"all"
+ },
+ "vpc_offering": {
+ "name": "VPC off",
+ "displaytext": "VPC off",
+ "supportedservices": "Dhcp,Dns,SourceNat,PortForwarding,Vpn,Lb,UserData,StaticNat,NetworkACL"
+ },
+ "vpc": {
+ "name": "TestVPC",
+ "displaytext": "TestVPC",
+ "cidr": "10.0.0.1/24"
+ },
+ "shared_network_offering": {
+ "name": "MySharedOffering",
+ "displaytext": "MySharedOffering",
+ "guestiptype": "Shared",
+ "supportedservices": "Dhcp,Dns,UserData",
+ "specifyVlan" : "False",
+ "specifyIpRanges" : "False",
+ "traffictype": "GUEST",
+ "serviceProviderList" : {
+ "Dhcp": "VirtualRouter",
+ "Dns": "VirtualRouter",
+ "UserData": "VirtualRouter"
+ }
+ },
+ "security_group" : { "name": "custom_Sec_Grp" },
+ "ingress_rule": {
+ "protocol": "TCP",
+ "startport": "22",
+ "endport": "22",
+ "cidrlist": "0.0.0.0/0"
+ },
+ "ostype": "CentOS 5.3 (64-bit)",
+ "sleep": 90,
+ "timeout": 10,
+ "advanced_sg": {
+ "zone": {
+ "name": "",
+ "dns1": "8.8.8.8",
+ "internaldns1": "192.168.100.1",
+ "networktype": "Advanced",
+ "securitygroupenabled": "true"
+ },
+ "securitygroupenabled": "true"
+ }
+}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/029f4e36/tools/marvin/marvin/configGenerator.py
----------------------------------------------------------------------
diff --git a/tools/marvin/marvin/configGenerator.py b/tools/marvin/marvin/configGenerator.py
index 6d5b70d..0d79e8e 100644
--- a/tools/marvin/marvin/configGenerator.py
+++ b/tools/marvin/marvin/configGenerator.py
@@ -330,7 +330,10 @@ class ConfigManager(object):
'''
def __init__(self):
- self.filePath = "config/config.cfg"
+ # Joining path with current directory will avoid relative path issue
+ # It will take correct path irrespective of from where the test case is run
+ dirPath = os.path.dirname(__file__)
+ self.filePath = os.path.join(dirPath, 'config/config.cfg')
self.parsedDict = None
if self.__verifyFile(self.filePath) is not False:
self.parsedDict = self.__parseConfig(self.filePath)
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/029f4e36/tools/marvin/marvin/integration/lib/base.py
----------------------------------------------------------------------
diff --git a/tools/marvin/marvin/integration/lib/base.py b/tools/marvin/marvin/integration/lib/base.py
index b43af33..aa23029 100755
--- a/tools/marvin/marvin/integration/lib/base.py
+++ b/tools/marvin/marvin/integration/lib/base.py
@@ -2262,6 +2262,8 @@ class Zone:
cmd.internaldns2 = services["internaldns2"]
if domainid:
cmd.domainid = domainid
+ if "securitygroupenabled" in services:
+ cmd.securitygroupenabled = services["securitygroupenabled"]
return Zone(apiclient.createZone(cmd).__dict__)
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/029f4e36/tools/marvin/marvin/integration/lib/common.py
----------------------------------------------------------------------
diff --git a/tools/marvin/marvin/integration/lib/common.py b/tools/marvin/marvin/integration/lib/common.py
index c1ae31b..c524675 100644
--- a/tools/marvin/marvin/integration/lib/common.py
+++ b/tools/marvin/marvin/integration/lib/common.py
@@ -799,35 +799,45 @@ def get_free_vlan(apiclient, zoneid):
assert len(list_physical_networks_response) > 0, "No physical networks found in zone %s" % zoneid
physical_network = list_physical_networks_response[0]
- vlans = xsplit(physical_network.vlan, ['-', ','])
- assert len(vlans) > 0
- assert int(vlans[0]) < int(vlans[-1]), "VLAN range %s was improperly split" % physical_network.vlan
-
- usedVlanIds = []
networks = list_networks(apiclient, zoneid= zoneid, type='Shared')
+ usedVlanIds = []
+
if isinstance(networks, list) and len(networks) > 0:
- usedVlanIds = [int(nw.vlan) for nw in networks]
+ usedVlanIds = [int(nw.vlan) for nw in networks if nw.vlan!="untagged"]
+
+ if hasattr(physical_network, "vlan") is False:
+ while True:
+ shared_ntwk_vlan = random.randrange(1,4095)
+ if shared_ntwk_vlan in usedVlanIds:
+ continue
+ else:
+ break
+ else:
+ vlans = xsplit(physical_network.vlan, ['-', ','])
+
+ assert len(vlans) > 0
+ assert int(vlans[0]) < int(vlans[-1]), "VLAN range %s was improperly split" % physical_network.vlan
- retriesCount = 20 #Assuming random function will give different integer each time
+ retriesCount = 20 #Assuming random function will give different integer each time
- shared_ntwk_vlan = None
+ shared_ntwk_vlan = None
- while True:
+ while True:
- if retriesCount == 0:
- break
+ if retriesCount == 0:
+ break
- free_vlan = int(vlans[-1]) + random.randrange(1, 20)
+ free_vlan = int(vlans[-1]) + random.randrange(1, 20)
- if free_vlan > 4095:
- free_vlan = int(vlans[0]) - random.randrange(1, 20)
- if free_vlan < 0 or (free_vlan in usedVlanIds):
- retriesCount -= 1
- continue
- else:
- shared_ntwk_vlan = free_vlan
- break
+ if free_vlan > 4095:
+ free_vlan = int(vlans[0]) - random.randrange(1, 20)
+ if free_vlan < 0 or (free_vlan in usedVlanIds):
+ retriesCount -= 1
+ continue
+ else:
+ shared_ntwk_vlan = free_vlan
+ break
return physical_network, shared_ntwk_vlan
[2/3] CLOUDSTACK-2237: Automation - Adding new test cases for
security group in advanced zone feature
Posted by gi...@apache.org.
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/029f4e36/test/integration/component/test_advancedsg_networks.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_advancedsg_networks.py b/test/integration/component/test_advancedsg_networks.py
index 4834351..7f3a390 100644
--- a/test/integration/component/test_advancedsg_networks.py
+++ b/test/integration/component/test_advancedsg_networks.py
@@ -17,167 +17,156 @@
""" P1 tests for networks in advanced zone with security groups
"""
-#Import Local Modules
-import marvin
-from marvin.cloudstackTestCase import *
-from marvin.cloudstackAPI import *
-from marvin.integration.lib.utils import *
-from marvin.integration.lib.base import *
-from marvin.integration.lib.common import *
-import netaddr
+from marvin.cloudstackTestCase import cloudstackTestCase, unittest
+from ddt import ddt, data
+from marvin.integration.lib.base import (Zone,
+ ServiceOffering,
+ Account,
+ NetworkOffering,
+ Network,
+ VirtualMachine,
+ Domain,
+ VpcOffering,
+ VPC,
+ SecurityGroup)
+
+from marvin.integration.lib.common import (get_domain,
+ get_zone,
+ get_template,
+ get_free_vlan,
+ list_virtual_machines,
+ wait_for_cleanup)
+
+from marvin.integration.lib.utils import (cleanup_resources,
+ random_gen,
+ validateList)
+from marvin.cloudstackAPI import (authorizeSecurityGroupIngress,
+ revokeSecurityGroupIngress)
from nose.plugins.attrib import attr
+from marvin.codes import PASS
+import time
+import random
+
+class TestCreateZoneSG(cloudstackTestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ cloudstackTestClient = super(TestCreateZoneSG,cls).getClsTestClient()
+ cls.api_client = cloudstackTestClient.getApiClient()
+
+ # Fill services from the external config file
+ cls.services = cloudstackTestClient.getConfigParser().parsedDict
+
+ # Get Zone, Domain and templates
+ cls.domain = get_domain(cls.api_client, cls.services)
+
+ cls._cleanup = []
+ return
+
+ @classmethod
+ def tearDownClass(cls):
+ try:
+ #Cleanup resources used
+ cleanup_resources(cls.api_client, cls._cleanup)
+ except Exception as e:
+ raise Exception("Warning: Exception during cleanup : %s" % e)
+ return
+
+ def setUp(self):
+ self.api_client = self.testClient.getApiClient()
+ self.debug(self.services)
+ self.cleanup = []
+ return
+
+ def tearDown(self):
+ try:
+ #Clean up, terminate the created network offerings
+ cleanup_resources(self.api_client, self.cleanup)
+ except Exception as e:
+ raise Exception("Warning: Exception during cleanup : %s" % e)
+
+ return
+
+ def createzone(self, sec_grps=True):
+
+ self.services["advanced_sg"]["zone"]["name"] = "AdvZoneSG-" + random_gen(size=6)
+ self.services["advanced_sg"]["zone"]["securitygroupenabled"] = sec_grps
+
+ self.debug(self.services["advanced_sg"]["zone"]["securitygroupenabled"])
+
+ try:
+ zone = Zone.create(self.api_client, self.services["advanced_sg"]["zone"])
+ self.cleanup.append(zone)
+ except Exception as e:
+ self.fail("Exception while creating zone: %s" % e)
+ return zone
+
+ def assert_on_sg_flag(self, flag, listzones):
+
+ self.assertEqual(listzones[0].securitygroupsenabled, flag,
+ "Security Group enabled flag is %s with created Zone"
+ % listzones[0].securitygroupsenabled)
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test_01_createZone_secGrp_enabled(self):
+ """ Test to verify Advance Zone with security group enabled can be created"""
+
+ # Validate:
+ # Create Advance Zone SG enabled using API
+
+ self.debug("Creating zone with secGrps Enabled")
+ zone = self.createzone()
+ self.debug("Created zone : %s" % zone.id)
+
+ listzones = Zone.list(self.api_client, id=zone.id)
+
+ self.debug("Checking if SecGroup flag is enabled for the zone")
+ self.assert_on_sg_flag(True, listzones)
+
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test_02_createZone_secGrp_disabled(self):
+ """ Test to verify Advance Zone created with flag
+ securitygroupsenabled=False"""
+
+ # Validate:
+ # Create Advance Zone without SG enabled
+ # Verify that the SG enabled flag is false
+
+ self.debug("Creating zone with secGrps Enabled")
+ zone = self.createzone(sec_grps=False)
+ self.debug("Created zone : %s" % zone.id)
+
+ listzones = Zone.list(self.api_client, id=zone.id)
+
+ self.debug("Checking if SecGroup flag is False for the zone")
+ self.assert_on_sg_flag(False, listzones)
+
+ return
-class Services:
- """ Test networks in advanced zone with security groups"""
-
- def __init__(self):
- self.services = {
- "domain": {
- "name": "DOM",
- },
- "project": {
- "name": "Project",
- "displaytext": "Test project",
- },
- "account": {
- "email": "admin-XABU1@test.com",
- "firstname": "admin-XABU1",
- "lastname": "admin-XABU1",
- "username": "admin-XABU1",
- # Random characters are appended for unique
- # username
- "password": "fr3sca",
- },
- "service_offering": {
- "name": "Tiny Instance",
- "displaytext": "Tiny Instance",
- "cpunumber": 1,
- "cpuspeed": 100, # in MHz
- "memory": 128, # In MBs
- },
- "shared_network_offering_sg": {
- "name": 'MySharedOffering-sg',
- "displaytext": 'MySharedOffering-sg',
- "guestiptype": 'Shared',
- "supportedservices": 'Dhcp,Dns,UserData,SecurityGroup',
- "specifyVlan" : "False",
- "specifyIpRanges" : "False",
- "traffictype": 'GUEST',
- "serviceProviderList" : {
- "Dhcp": 'VirtualRouter',
- "Dns": 'VirtualRouter',
- "UserData": 'VirtualRouter',
- "SecurityGroup": 'SecurityGroupProvider'
- },
- },
- "shared_network_offering": {
- "name": 'MySharedOffering',
- "displaytext": 'MySharedOffering',
- "guestiptype": 'Shared',
- "supportedservices": 'Dhcp,Dns,UserData',
- "specifyVlan" : "False",
- "specifyIpRanges" : "False",
- "traffictype": 'GUEST',
- "serviceProviderList" : {
- "Dhcp": 'VirtualRouter',
- "Dns": 'VirtualRouter',
- "UserData": 'VirtualRouter'
- },
- },
- "shared_network_sg": {
- "name": "MyIsolatedNetwork - Test",
- "displaytext": "MyIsolatedNetwork",
- "networkofferingid":"1",
- "vlan" :1200,
- "gateway" :"172.16.15.1",
- "netmask" :"255.255.255.0",
- "startip" :"172.16.15.2",
- "endip" :"172.16.15.20",
- "acltype" : "Domain",
- "scope":"all",
- },
- "shared_network": {
- "name": "MySharedNetwork - Test",
- "displaytext": "MySharedNetwork",
- "vlan" :1201,
- "gateway" :"172.16.15.1",
- "netmask" :"255.255.255.0",
- "startip" :"172.16.15.21",
- "endip" :"172.16.15.41",
- "acltype" : "Domain",
- "scope":"all",
- },
- "isolated_network_offering": {
- "name": 'Network offering-DA services',
- "displaytext": 'Network offering-DA services',
- "guestiptype": 'Isolated',
- "supportedservices": 'Dhcp,Dns,SourceNat,PortForwarding,Vpn,Firewall,Lb,UserData,StaticNat',
- "traffictype": 'GUEST',
- "availability": 'Optional',
- "serviceProviderList": {
- "Dhcp": 'VirtualRouter',
- "Dns": 'VirtualRouter',
- "SourceNat": 'VirtualRouter',
- "PortForwarding": 'VirtualRouter',
- "Vpn": 'VirtualRouter',
- "Firewall": 'VirtualRouter',
- "Lb": 'VirtualRouter',
- "UserData": 'VirtualRouter',
- "StaticNat": 'VirtualRouter',
- },
- },
- "isolated_network": {
- "name": "Isolated Network",
- "displaytext": "Isolated Network",
- },
- "virtual_machine": {
- "displayname": "Test VM",
- "username": "root",
- "password": "password",
- "ssh_port": 22,
- "hypervisor": 'XenServer',
- # Hypervisor type should be same as
- # hypervisor type of cluster
- "privateport": 22,
- "publicport": 22,
- "protocol": 'TCP',
- },
- "ostype": 'CentOS 5.3 (64-bit)',
- # Cent OS 5.3 (64 bit)
- "sleep": 90,
- "timeout": 10,
- "mode": 'advanced',
- "securitygroupenabled": 'true'
- }
-
class TestNetworksInAdvancedSG(cloudstackTestCase):
-
+ """Test Creation of different types of networks in SG enabled advanced zone"""
+
@classmethod
def setUpClass(cls):
- cls.api_client = super(
- TestNetworksInAdvancedSG,
- cls
- ).getClsTestClient().getApiClient()
-
- cls.services = Services().services
+ cloudstackTestClient = super(TestNetworksInAdvancedSG,cls).getClsTestClient()
+ cls.api_client = cloudstackTestClient.getApiClient()
+
+ cls.services = cloudstackTestClient.getConfigParser().parsedDict
# Get Zone, Domain and templates
cls.domain = get_domain(cls.api_client, cls.services)
cls.zone = get_zone(cls.api_client, cls.services)
- cls.template = get_template(
- cls.api_client,
- cls.zone.id,
- cls.services["ostype"]
- )
-
+ cls.template = get_template(cls.api_client, cls.zone.id,
+ cls.services["ostype"])
+
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
cls.services["virtual_machine"]["template"] = cls.template.id
- cls.service_offering = ServiceOffering.create(
- cls.api_client,
- cls.services["service_offering"]
- )
-
+ cls.service_offering = ServiceOffering.create(cls.api_client,cls.services["service_offering"])
+
cls._cleanup = [
cls.service_offering,
]
@@ -201,521 +190,236 @@ class TestNetworksInAdvancedSG(cloudstackTestCase):
self.cleanup_domains = []
self.cleanup_projects = []
self.cleanup_vms = []
+ self.cleanup_nwOfferings = []
+
+ self.services["shared_network_offering_sg"]["specifyVlan"] = "True"
+ self.services["shared_network_offering_sg"]["specifyIpRanges"] = "True"
+
+ #Create Network Offering
+ self.shared_network_offering_sg = NetworkOffering.create(self.api_client, self.services["shared_network_offering_sg"],
+ conservemode=False)
+
+ self.cleanup_nwOfferings.append(self.shared_network_offering_sg)
+
+ #Update network offering state from disabled to enabled.
+ NetworkOffering.update(self.shared_network_offering_sg,self.api_client,state="enabled")
return
def tearDown(self):
+ # all exceptions during cleanup will be appended to this list
+ exceptions = []
try:
#Clean up, terminate the created network offerings
cleanup_resources(self.api_client, self.cleanup)
except Exception as e:
- raise Exception("Warning: Exception during cleanup : %s" % e)
-
+ self.debug("Warning: Exception during cleanup : %s" % e)
+ exceptions.append(e)
+
#below components is not a part of cleanup because to mandate the order and to cleanup network
try:
for vm in self.cleanup_vms:
- vm.delete(self.api_client)
+ vm.delete(self.api_client)
except Exception as e:
- raise Exception("Warning: Exception during virtual machines cleanup : %s" % e)
-
+ self.debug("Warning: Exception during virtual machines cleanup : %s" % e)
+ exceptions.append(e)
+
+ # Wait for VMs to expunge
+ wait_for_cleanup(self.api_client, ["expunge.delay", "expunge.interval"])
+
try:
for project in self.cleanup_projects:
- project.delete(self.api_client)
+ project.delete(self.api_client)
except Exception as e:
- raise Exception("Warning: Exception during project cleanup : %s" % e)
-
+ self.debug("Warning: Exception during project cleanup : %s" % e)
+ exceptions.append(e)
+
try:
for account in self.cleanup_accounts:
- account.delete(self.api_client)
- except Exception as e:
- raise Exception("Warning: Exception during account cleanup : %s" % e)
-
- try:
- for domain in self.cleanup_domains:
- domain.delete(self.api_client)
+ account.delete(self.api_client)
except Exception as e:
- raise Exception("Warning: Exception during domain cleanup : %s" % e)
-
+ self.debug("Warning: Exception during account cleanup : %s" % e)
+ exceptions.append(e)
+
#Wait till all resources created are cleaned up completely and then attempt to delete Network
time.sleep(self.services["sleep"])
-
+
try:
for network in self.cleanup_networks:
network.delete(self.api_client)
except Exception as e:
- raise Exception("Warning: Exception during network cleanup : %s" % e)
+ self.debug("Warning: Exception during network cleanup : %s" % e)
+ exceptions.append(e)
+
+ try:
+ for domain in self.cleanup_domains:
+ domain.delete(self.api_client)
+ except Exception as e:
+ self.debug("Warning: Exception during domain cleanup : %s" % e)
+ exceptions.append(e)
+
+ try:
+ for network_offering in self.cleanup_nwOfferings:
+ network_offering.update(self.api_client, state="disabled")
+ network_offering.delete(self.api_client)
+ except Exception as e:
+ self.debug("Warning: Exception during network cleanup : %s" % e)
+ exceptions.append(e)
+
+ if len(exceptions) > 0:
+ self.fail("There were exceptions during cleanup: %s" % exceptions)
+ return
+
+ def setSharedNetworkParams(self, network, range=20):
+
+ # @range: range decides the endip. Pass the range as "x" if you want the difference between the startip
+ # and endip as "x"
+ # Set the subnet number of shared networks randomly prior to execution
+ # of each test case to avoid overlapping of ip addresses
+ shared_network_subnet_number = random.randrange(1,254)
+
+ self.services[network]["gateway"] = "172.16."+str(shared_network_subnet_number)+".1"
+ self.services[network]["startip"] = "172.16."+str(shared_network_subnet_number)+".2"
+ self.services[network]["endip"] = "172.16."+str(shared_network_subnet_number)+"."+str(range+1)
+ self.services[network]["netmask"] = "255.255.255.0"
+
return
@attr(tags = ["advancedsg"])
- def test_createIsolatedNetwork(self):
+ def test_03_createIsolatedNetwork(self):
""" Test Isolated Network """
-
+
# Steps,
- # 1. create an Admin Account - admin-XABU1
- # 2. listPhysicalNetworks in available zone
- # 3. createNetworkOffering:
- # 4. Enable network offering - updateNetworkOffering - state=Enabled
- # 5. createNetwork
+ # 1. create Isolated Network Offering
+ # 2. Enable network offering - updateNetworkOffering - state=Enabled
+ # 3. Try to create Isolated Network
# Validations,
- # 1. listAccounts name=admin-XABU1, state=enabled returns your account
- # 2. listPhysicalNetworks should return at least one active physical network
- # 4. listNetworkOfferings - name=myisolatedoffering, should list enabled offering
- # 5. network creation should FAIL since isolated network is not supported in advanced zone with security groups.
-
- #Create admin account
- self.admin_account = Account.create(
- self.api_client,
- self.services["account"],
- admin=True,
- domainid=self.domain.id
- )
-
- self.cleanup_accounts.append(self.admin_account)
-
- #verify that the account got created with state enabled
- list_accounts_response = Account.list(
- self.api_client,
- id=self.admin_account.id,
- listall=True
- )
- self.assertEqual(
- isinstance(list_accounts_response, list),
- True,
- "listAccounts returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_accounts_response),
- 0,
- "listAccounts returned empty list."
- )
- self.assertEqual(
- list_accounts_response[0].state,
- "enabled",
- "The admin account created is not enabled."
- )
-
- self.debug("Admin type account created: %s" % self.admin_account.name)
-
- #Create an user account
- self.user_account = Account.create(
- self.api_client,
- self.services["account"],
- admin=False,
- domainid=self.domain.id
- )
-
- self.cleanup_accounts.append(self.user_account)
-
- #verify that the account got created with state enabled
- list_accounts_response = Account.list(
- self.api_client,
- id=self.user_account.id,
- listall=True
- )
- self.assertEqual(
- isinstance(list_accounts_response, list),
- True,
- "listAccounts returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_accounts_response),
- 0,
- "listAccounts returned empty list."
- )
- self.assertEqual(
- list_accounts_response[0].state,
- "enabled",
- "The user account created is not enabled."
- )
-
- self.debug("User type account created: %s" % self.user_account.name)
-
- #Verify that there should be at least one physical network present in zone.
- list_physical_networks_response = PhysicalNetwork.list(
- self.api_client,
- zoneid=self.zone.id
- )
- self.assertEqual(
- isinstance(list_physical_networks_response, list),
- True,
- "listPhysicalNetworks returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_physical_networks_response),
- 0,
- "listPhysicalNetworks should return at least one physical network."
- )
-
- physical_network = list_physical_networks_response[0]
-
- self.debug("Physical network found: %s" % physical_network.id)
-
+ # 1. Network creation should FAIL since isolated network is not supported in advanced zone with security groups.
+
#Create Network Offering
- self.isolated_network_offering = NetworkOffering.create(
- self.api_client,
- self.services["isolated_network_offering"],
- conservemode=False
- )
-
+
+ self.debug("Creating Isolated network offering")
+ self.isolated_network_offering = NetworkOffering.create(self.api_client, self.services["isolated_network_offering"],
+ conservemode=False)
+
self.cleanup.append(self.isolated_network_offering)
-
- #Verify that the network offering got created
- list_network_offerings_response = NetworkOffering.list(
- self.api_client,
- id=self.isolated_network_offering.id
- )
- self.assertEqual(
- isinstance(list_network_offerings_response, list),
- True,
- "listNetworkOfferings returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_network_offerings_response),
- 0,
- "listNetworkOfferings returned empty list."
- )
- self.assertEqual(
- list_network_offerings_response[0].state,
- "Disabled",
- "The network offering created should be bydefault disabled."
- )
-
+
self.debug("Isolated Network offering created: %s" % self.isolated_network_offering.id)
-
+
#Update network offering state from disabled to enabled.
- network_offering_update_response = NetworkOffering.update(
- self.isolated_network_offering,
- self.api_client,
- id=self.isolated_network_offering.id,
- state="enabled"
- )
-
- #Verify that the state of the network offering is updated
- list_network_offerings_response = NetworkOffering.list(
- self.api_client,
- id=self.isolated_network_offering.id
- )
- self.assertEqual(
- isinstance(list_network_offerings_response, list),
- True,
- "listNetworkOfferings returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_network_offerings_response),
- 0,
- "listNetworkOfferings returned empty list."
- )
- self.assertEqual(
- list_network_offerings_response[0].state,
- "Enabled",
- "The network offering state should get updated to Enabled."
- )
-
+ self.isolated_network_offering.update( self.api_client, state="enabled")
+
#create network using the isolated network offering created
try:
- self.isolated_network = Network.create(
- self.api_client,
- self.services["isolated_network"],
- networkofferingid=self.isolated_network_offering.id,
- zoneid=self.zone.id,
- )
- self.cleanup_networks.append(self.isolated_network)
+ self.debug("Trying to create Isolated network, this should fail")
+ self.isolated_network = Network.create(self.api_client, self.services["isolated_network"],
+ networkofferingid=self.isolated_network_offering.id,
+ zoneid=self.zone.id)
+ self.cleanup_networks.append(self.isolated_network)
self.fail("Create isolated network is invalid in advanced zone with security groups.")
except Exception as e:
- self.debug("Network creation failed because create isolated network is invalid in advanced zone with security groups.")
+ self.debug("Network creation failed because creating isolated network is invalid in advanced zone with security groups.\
+ Exception: %s" % e)
+ return
@attr(tags = ["advancedsg"])
- def test_createSharedNetwork_withoutSG(self):
- """ Test Shared Network with without SecurityProvider """
-
+ def test_04_createSharedNetwork_withoutSG(self):
+ """ Test Shared Network creation without Security Group Service Provider in Network Offering"""
+
# Steps,
- # 1. create an Admin account
- # 2. create a shared NetworkOffering
- # 3. enable the network offering
- # 4. listPhysicalNetworks
- # 5. createNetwork
+ # 1. Create a shared Network offering without SG
+ # 2. Enable the network offering
+ # 3. Try to create shared network using the above offering
# Validations,
- # 1. listAccounts state=enabled returns your account
- # 2. listNetworkOfferings - name=mysharedoffering , should list offering in disabled state
- # 3. listNetworkOfferings - name=mysharedoffering, should list enabled offering
- # 4. listPhysicalNetworks should return at least one active physical network
- # 5. network creation should FAIL since there is no SecurityProvide in the network offering
-
- #Create admin account
- self.admin_account = Account.create(
- self.api_client,
- self.services["account"],
- admin=True,
- domainid=self.domain.id
- )
-
- self.cleanup_accounts.append(self.admin_account)
-
- #verify that the account got created with state enabled
- list_accounts_response = Account.list(
- self.api_client,
- id=self.admin_account.id,
- listall=True
- )
- self.assertEqual(
- isinstance(list_accounts_response, list),
- True,
- "listAccounts returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_accounts_response),
- 0,
- "listAccounts returned empty list."
- )
- self.assertEqual(
- list_accounts_response[0].state,
- "enabled",
- "The admin account created is not enabled."
- )
-
- self.debug("Domain admin account created: %s" % self.admin_account.id)
-
- #Verify that there should be at least one physical network present in zone.
- list_physical_networks_response = PhysicalNetwork.list(
- self.api_client,
- zoneid=self.zone.id
- )
- self.assertEqual(
- isinstance(list_physical_networks_response, list),
- True,
- "listPhysicalNetworks returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_physical_networks_response),
- 0,
- "listPhysicalNetworks should return at least one physical network."
- )
-
- physical_network = list_physical_networks_response[0]
-
- self.debug("Physical Network found: %s" % physical_network.id)
-
+ # 1. Network creation should FAIL since there is no SecurityProvider in the network offering
+
self.services["shared_network_offering"]["specifyVlan"] = "True"
self.services["shared_network_offering"]["specifyIpRanges"] = "True"
-
+
#Create Network Offering
- self.shared_network_offering = NetworkOffering.create(
- self.api_client,
- self.services["shared_network_offering"],
- conservemode=False
- )
-
- self.cleanup.append(self.shared_network_offering)
-
- #Verify that the network offering got created
- list_network_offerings_response = NetworkOffering.list(
- self.api_client,
- id=self.shared_network_offering.id
- )
- self.assertEqual(
- isinstance(list_network_offerings_response, list),
- True,
- "listNetworkOfferings returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_network_offerings_response),
- 0,
- "listNetworkOfferings returned empty list."
- )
- self.assertEqual(
- list_network_offerings_response[0].state,
- "Disabled",
- "The network offering created should be bydefault disabled."
- )
-
- self.debug("Shared Network Offering created: %s" % self.shared_network_offering.id)
-
+ self.shared_network_offering = NetworkOffering.create(self.api_client, self.services["shared_network_offering"],
+ conservemode=False)
+
+ self.cleanup_nwOfferings.append(self.shared_network_offering)
+
#Update network offering state from disabled to enabled.
- network_offering_update_response = NetworkOffering.update(
- self.shared_network_offering,
- self.api_client,
- id=self.shared_network_offering.id,
- state="enabled"
- )
-
- #Verify that the state of the network offering is updated
- list_network_offerings_response = NetworkOffering.list(
- self.api_client,
- id=self.shared_network_offering.id
- )
- self.assertEqual(
- isinstance(list_network_offerings_response, list),
- True,
- "listNetworkOfferings returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_network_offerings_response),
- 0,
- "listNetworkOfferings returned empty list."
- )
- self.assertEqual(
- list_network_offerings_response[0].state,
- "Enabled",
- "The network offering state should get updated to Enabled."
- )
-
+ NetworkOffering.update(self.shared_network_offering,self.api_client,state="enabled")
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
#create network using the shared network offering created
self.services["shared_network"]["acltype"] = "domain"
+ self.services["shared_network"]["vlan"] = vlan
self.services["shared_network"]["networkofferingid"] = self.shared_network_offering.id
self.services["shared_network"]["physicalnetworkid"] = physical_network.id
-
+
+ self.setSharedNetworkParams("shared_network")
+
try:
- self.shared_network = Network.create(
- self.api_client,
- self.services["shared_network"],
- networkofferingid=self.shared_network_offering.id,
- zoneid=self.zone.id
- )
+ self.shared_network = Network.create(self.api_client, self.services["shared_network"],
+ networkofferingid=self.shared_network_offering.id, zoneid=self.zone.id)
self.cleanup_networks.append(self.shared_network)
self.fail("Network created without SecurityProvider , which is invalid")
except Exception as e:
- self.debug("Network creation failed because there is no SecurityProvider in the network offering.")
-
+ self.debug("Network creation failed because there is no SecurityProvider in the network offering.\
+ Exception: %s" % e)
+ return
+
@attr(tags = ["advancedsg"])
- def test_deployVM_SharedwithSG(self):
- """ Test VM deployment in shared networks with SecurityProvider """
-
+ def test_05_deployVM_SharedwithSG(self):
+ """ Test VM deployment in shared networks with SecurityGroup Provider """
+
# Steps,
- # 0. create a user account
- # 1. Create one shared Network (scope=ALL, different IP ranges)
- # 2. deployVirtualMachine in the above networkid within the user account
- # 3. delete the user account
+ # 1. create an account
+ # 2. Create one shared Network with sec group
+ # 3. deployVirtualMachine in the above networkid within the user account
+ # 4. delete the user account
# Validations,
# 1. shared network should be created successfully
# 2. VM should deploy successfully
-
+
#Create admin account
- self.admin_account = Account.create(
- self.api_client,
- self.services["account"],
- admin=True,
- domainid=self.domain.id
- )
-
+ self.admin_account = Account.create(self.api_client, self.services["account"], admin=True,
+ domainid=self.domain.id)
+
self.cleanup_accounts.append(self.admin_account)
-
- #verify that the account got created with state enabled
- list_accounts_response = Account.list(
- self.api_client,
- id=self.admin_account.id,
- liistall=True
- )
- self.assertEqual(
- isinstance(list_accounts_response, list),
- True,
- "listAccounts returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_accounts_response),
- 0,
- "listAccounts returned empty list."
- )
- self.assertEqual(
- list_accounts_response[0].state,
- "enabled",
- "The admin account created is not enabled."
- )
-
+
self.debug("Admin type account created: %s" % self.admin_account.name)
-
+
self.services["shared_network_offering_sg"]["specifyVlan"] = "True"
self.services["shared_network_offering_sg"]["specifyIpRanges"] = "True"
-
+
#Create Network Offering
- self.shared_network_offering_sg = NetworkOffering.create(
- self.api_client,
- self.services["shared_network_offering_sg"],
- conservemode=False
- )
-
- self.cleanup.append(self.shared_network_offering_sg)
-
- #Verify that the network offering got created
- list_network_offerings_response = NetworkOffering.list(
- self.api_client,
- id=self.shared_network_offering_sg.id
- )
- self.assertEqual(
- isinstance(list_network_offerings_response, list),
- True,
- "listNetworkOfferings returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_network_offerings_response),
- 0,
- "listNetworkOfferings returned empty list."
- )
- self.assertEqual(
- list_network_offerings_response[0].state,
- "Disabled",
- "The network offering created should be bydefault disabled."
- )
-
+ self.shared_network_offering_sg = NetworkOffering.create(self.api_client,self.services["shared_network_offering_sg"],
+ conservemode=False)
+
+ self.cleanup_nwOfferings.append(self.shared_network_offering_sg)
+
self.debug("Shared Network offering created: %s" % self.shared_network_offering_sg.id)
-
+
#Update network offering state from disabled to enabled.
- network_offering_update_response = NetworkOffering.update(
- self.shared_network_offering_sg,
- self.api_client,
- id=self.shared_network_offering_sg.id,
- state="enabled"
- )
-
- #Verify that the state of the network offering is updated
- list_network_offerings_response = NetworkOffering.list(
- self.api_client,
- id=self.shared_network_offering_sg.id
- )
- self.assertEqual(
- isinstance(list_network_offerings_response, list),
- True,
- "listNetworkOfferings returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_network_offerings_response),
- 0,
- "listNetworkOfferings returned empty list."
- )
- self.assertEqual(
- list_network_offerings_response[0].state,
- "Enabled",
- "The network offering state should get updated to Enabled."
- )
-
- physical_network = PhysicalNetwork.list(self.api_client)[0]
+ self.shared_network_offering_sg.update(self.api_client,state="enabled")
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
#create network using the shared network offering created
self.services["shared_network_sg"]["acltype"] = "domain"
+ self.services["shared_network_sg"]["vlan"] = vlan
self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
- self.shared_network_sg = Network.create(
- self.api_client,
- self.services["shared_network_sg"],
- domainid=self.admin_account.domainid,
- networkofferingid=self.shared_network_offering_sg.id,
- zoneid=self.zone.id
- )
-
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ self.shared_network_sg = Network.create(self.api_client, self.services["shared_network_sg"], domainid=self.admin_account.domainid,
+ networkofferingid=self.shared_network_offering_sg.id, zoneid=self.zone.id)
+
self.cleanup_networks.append(self.shared_network_sg)
-
+
list_networks_response = Network.list(
self.api_client,
id=self.shared_network_sg.id
)
- self.assertEqual(
- isinstance(list_networks_response, list),
- True,
- "listNetworks returned invalid object in response."
- )
- self.assertNotEqual(
- len(list_networks_response),
- 0,
- "listNetworks returned empty list."
- )
+ self.assertEqual(validateList(list_networks_response)[0], PASS, "Networks list validation failed, list is %s" %
+ list_networks_response)
self.assertEqual(
list_networks_response[0].specifyipranges,
True,
@@ -723,34 +427,1514 @@ class TestNetworksInAdvancedSG(cloudstackTestCase):
)
self.debug("Shared Network created: %s" % self.shared_network_sg.id)
-
- self.shared_network_admin_account_virtual_machine = VirtualMachine.create(
- self.api_client,
- self.services["virtual_machine"],
- accountid=self.admin_account.name,
- domainid=self.admin_account.domainid,
- networkids=self.shared_network_sg.id,
- serviceofferingid=self.service_offering.id
- )
- self.cleanup_vms.append(self.shared_network_admin_account_virtual_machine)
- vms = VirtualMachine.list(
- self.api_client,
- id=self.shared_network_admin_account_virtual_machine.id,
- listall=True
- )
- self.assertEqual(
- isinstance(vms, list),
- True,
- "listVirtualMachines returned invalid object in response."
- )
- self.assertNotEqual(
- len(vms),
- 0,
- "listVirtualMachines returned empty list."
- )
- self.debug("Virtual Machine created: %s" % self.shared_network_admin_account_virtual_machine.id)
-
- ip_range = list(netaddr.iter_iprange(unicode(self.services["shared_network_sg"]["startip"]), unicode(self.services["shared_network_sg"]["endip"])))
- if netaddr.IPAddress(unicode(vms[0].nic[0].ipaddress)) not in ip_range:
- self.fail("Virtual machine ip should be from the ip range assigned to network created.")
+ try:
+ virtual_machine = VirtualMachine.create(self.api_client,self.services["virtual_machine"],accountid=self.admin_account.name,
+ domainid=self.admin_account.domainid, networkids=[self.shared_network_sg.id],
+ serviceofferingid=self.service_offering.id
+ )
+ except Exception as e:
+ self.fail("Exception while deploying virtual machine: %s" % e)
+ self.cleanup_vms.append(virtual_machine)
+
+ vms = VirtualMachine.list(self.api_client, id=virtual_machine.id,listall=True)
+
+ self.assertEqual(validateList(vms)[0], PASS, "vms list validation failed, list is %s" % vms)
+ self.debug("Virtual Machine created: %s" % virtual_machine.id)
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test_06_SharedNwSGAccountSpecific(self):
+ """ Test Account specific shared network creation with SG"""
+
+ # Steps,
+ # 1. create a user account
+ # 2. Create one shared Network (scope=Account) specific to this account
+
+ # Validations,
+ # 1. shared network should be created successfully
+
+ #Create admin account
+ self.admin_account = Account.create(self.api_client,self.services["account"],admin=True,
+ domainid=self.domain.id)
+
+ self.cleanup_accounts.append(self.admin_account)
+
+ #Create user account
+ self.user_account = Account.create(self.api_client,self.services["account"],domainid=self.domain.id)
+
+ self.cleanup_accounts.append(self.user_account)
+
+ self.services["shared_network_offering_sg"]["specifyVlan"] = "True"
+ self.services["shared_network_offering_sg"]["specifyIpRanges"] = "True"
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["acltype"] = "account"
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ try:
+ self.shared_network_sg_admin_account = Network.create(self.api_client, self.services["shared_network_sg"],
+ accountid=self.admin_account.name, domainid=self.admin_account.domainid,
+ networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id)
+ except Exception as e:
+ self.fail("Exception while creating account specific shared network: %s" % e)
+
+ self.debug("Created shared network %s with admin account" % self.shared_network_sg_admin_account.id)
+
+ list_networks_response = Network.list(
+ self.api_client,
+ id=self.shared_network_sg_admin_account.id
+ )
+ self.assertEqual(validateList(list_networks_response)[0], PASS, "networks list validation failed, list is %s" %
+ list_networks_response)
+
+ self.debug("Shared Network created: %s" % self.shared_network_sg_admin_account.id)
+
+ self.debug("Creating shared account in user account: %s" % self.user_account.id)
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["vlan"] = vlan
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ try:
+ self.shared_network_sg_user_account = Network.create(self.api_client,self.services["shared_network_sg"],
+ accountid=self.user_account.name,domainid=self.user_account.domainid,
+ networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id)
+ except Exception as e:
+ self.fail("Exception while creating account specific shared network: %s" % e)
+
+ self.debug("Created shared network %s with user account" % self.shared_network_sg_user_account.id)
+
+ list_networks_response = Network.list(self.api_client,id=self.shared_network_sg_user_account.id)
+
+ self.assertEqual(validateList(list_networks_response)[0], PASS, "networks list validation failed, list is %s" %
+ list_networks_response)
+
+ self.debug("Shared Network created: %s" % self.shared_network_sg_user_account.id)
+
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test_07_SharedNwSG_DomainWide_SubdomainAcccess(self):
+ """ Test Domain wide shared network with SG, with subdomain access set True"""
+
+ # Steps,
+ # 1. create a Domain and subdomain
+ # 2. Create one shared Network in parent domain with SG and set subdomain access True
+ # 3. Deploy a VM in subdomain using the shared network
+
+ # Validations,
+ # 1. shared network should be created successfully
+ # 2. Shared network should be able to be accessed within subdomain (VM should be deployed)
+
+ #Create Domain
+ self.debug("Creating parent domain")
+ self.parent_domain = Domain.create(self.api_client, services=self.services["domain"],
+ parentdomainid=self.domain.id)
+
+ self.debug("Created domain %s" % self.parent_domain.id)
+ self.debug("Creating child domain under this domain")
+ self.child_domain = Domain.create(self.api_client,services=self.services["domain"],
+ parentdomainid=self.parent_domain)
+
+ self.debug("Created child domain: %s" % self.child_domain.id)
+
+ self.services["shared_network_offering_sg"]["specifyVlan"] = "True"
+ self.services["shared_network_offering_sg"]["specifyIpRanges"] = "True"
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["acltype"] = "domain"
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ try:
+ self.shared_network_sg = Network.create(self.api_client,self.services["shared_network_sg"],
+ domainid=self.parent_domain.id,networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id,subdomainaccess=True)
+ except Exception as e:
+ self.fail("Exception whle creating domain wide shared network: %s" % e)
+
+ self.debug("Created shared network: %s" % self.shared_network_sg.id)
+
+ self.cleanup_networks.append(self.shared_network_sg)
+
+ list_networks_response = Network.list(self.api_client,id=self.shared_network_sg.id,listall=True)
+
+ self.debug("network response: %s" % list_networks_response)
+ self.assertEqual(validateList(list_networks_response)[0], PASS, "networks list validation failed")
+
+ self.debug("Shared Network created: %s" % self.shared_network_sg.id)
+
+ try:
+ virtual_machine = VirtualMachine.create(self.api_client, self.services["virtual_machine"],
+ domainid=self.child_domain.id,networkids=[self.shared_network_sg.id],
+ serviceofferingid=self.service_offering.id)
+ except Exception as e:
+ self.fail("Exception while deploying VM in domain wide shared network: %s" % e)
+
+ self.debug("Created virtual machine %s within the shared network" % virtual_machine.id)
+
+ self.cleanup_vms.append(virtual_machine)
+
+ return
+
+ @unittest.skip("Skip - Failing - WIP")
+ @attr(tags = ["advancedsg"])
+ def test_08_SharedNwSGAccountSpecific_samevlan_samesubnet(self):
+ """ Test Account specific shared network creation with SG in multiple accounts
+ with same subnet and vlan"""
+
+ # Steps,
+ # 1. create two different accouts
+ # 2. create account specific shared networks in both accounts with same subnet and vlan id
+
+ # Validations,
+ # 1. shared networks should be created successfully
+
+ #Create domain 1
+
+ domain1 = Domain.create(self.api_client,services=self.services["domain"],parentdomainid=self.domain.id)
+
+ self.debug("Created domain: %s" % domain1.id)
+ self.cleanup_domains.append(domain1)
+
+ account1 = Account.create(self.api_client,self.services["account"],domainid=domain1.id)
+
+ self.debug("Created account %s under domain %s" % (account1.id, domain1.id))
+ self.cleanup_accounts.append(account1)
+
+ domain2 = Domain.create(self.api_client, services=self.services["domain"],parentdomainid=self.domain.id)
+
+ self.debug("Created domain %s" % domain2.id)
+ self.cleanup_domains.append(domain1)
+
+ account2 = Account.create(self.api_client,self.services["account"],domainid=domain2.id)
+
+ self.debug("Created account %s under domain %s" % (account2.id, domain2.id))
+ self.cleanup_accounts.append(account1)
+
+ self.services["shared_network_offering_sg"]["specifyVlan"] = "True"
+ self.services["shared_network_offering_sg"]["specifyIpRanges"] = "True"
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["acltype"] = "account"
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ self.debug("Creating shared network in account 1: %s" % account1.name)
+ self.shared_network_sg_account1 = Network.create(self.api_client,self.services["shared_network_sg"],
+ accountid=account1.name,domainid=account1.domainid,
+ networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id)
+
+ self.debug("Created shared network: %s" % self.shared_network_sg_account1.id)
+
+ self.debug("Creating shared network in account 2 with same subnet and vlan: %s" % account1.name)
+
+ try:
+ self.shared_network_sg_account2 = Network.create(self.api_client,self.services["shared_network_sg"],
+ accountid=account2.name,domainid=account2.domainid,
+ networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id)
+ except Exception as e:
+ self.fail("Exception while creating account specific shared network with same subnet and vlan: %s" % e)
+
+ self.debug("Created shared network: %s" % self.shared_network_sg_account2.id)
+
+ return
+
+ @unittest.skip("Skip - Failing - WIP")
+ @attr(tags = ["advancedsg"])
+ def test_09_SharedNwDomainWide_samevlan_samesubnet(self):
+ """ Test Domain wide shared network creation with SG in different domains
+ with same vlan and same subnet"""
+
+ # Steps,
+ # 1. create two different domains
+ # 2. Create domain specific shared networks in both the domains using same subnet and vlan id
+
+ # Validations,
+ # 1. Shared networks should be created successfully
+
+ #Create domain 1
+
+ domain1 = Domain.create(self.api_client,services=self.services["domain"],parentdomainid=self.domain.id)
+
+ self.debug("Created domain: %s" % domain1.id)
+ self.cleanup_domains.append(domain1)
+
+ account1 = Account.create(self.api_client,self.services["account"],
+ domainid=domain1.id)
+
+ self.debug("Created account %s under domain %s" % (account1.id, domain1.id))
+ self.cleanup_accounts.append(account1)
+
+ domain2 = Domain.create(self.api_client,services=self.services["domain"],parentdomainid=self.domain.id)
+
+ self.debug("Created domain %s" % domain2.id)
+ self.cleanup_domains.append(domain2)
+
+ account2 = Account.create(self.api_client,self.services["account"],domainid=domain2.id)
+
+ self.debug("Created account %s under domain %s" % (account2.id, domain2.id))
+ self.cleanup_accounts.append(account2)
+
+ self.services["shared_network_offering_sg"]["specifyVlan"] = "True"
+ self.services["shared_network_offering_sg"]["specifyIpRanges"] = "True"
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["acltype"] = "domain"
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ self.debug("Creating shared network domain 1: %s" % domain1.name)
+ self.shared_network_sg_domain1 = Network.create(self.api_client,self.services["shared_network_sg"],
+ domainid=domain1.id,networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id, subdomainaccess=True)
+
+ self.debug("Created shared network: %s" % self.shared_network_sg_domain1.id)
+
+ list_networks_response = Network.list(self.api_client,id=self.shared_network_sg_domain1.id,listall=True)
+
+ self.assertEqual(validateList(list_networks_response)[0], PASS, "networks list validation failed, list: %s" %
+ list_networks_response)
+
+ self.debug("Creating shared network in domain 2 with same subnet and vlan: %s" % domain2.name)
+
+ try:
+ self.shared_network_sg_domain2 = Network.create(self.api_client,self.services["shared_network_sg"],
+ domainid=domain2.id,networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id,subdomainaccess=True)
+ except Exception as e:
+ self.fail("Exception while creating domain wide shared network with same subnet and vlan: %s" % e)
+
+ self.debug("Created shared network: %s" % self.shared_network_sg_domain2.id)
+
+ list_networks_response = Network.list(self.api_client,id=self.shared_network_sg_domain2.id)
+
+ self.assertEqual(validateList(list_networks_response)[0], PASS, "networks list validation failed, list: %s" %
+ list_networks_response)
+
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test_10_deleteSharedNwSGAccountSpecific_InUse(self):
+ """ Test delete Account specific shared network creation with SG which is in use"""
+
+ # Steps,
+ # 1. create a user account
+ # 2. Create account specific sg enabled shared network
+ # 3. Deply vm in the shared network
+ # 4. Try to delete shared network while it's stil in use by the vm
+
+ # Validations,
+ # 1. shared network deletion should fail
+
+ #Create admin account
+ self.admin_account = Account.create(self.api_client,self.services["account"],admin=True,
+ domainid=self.domain.id)
+
+ self.cleanup_accounts.append(self.admin_account)
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["acltype"] = "account"
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ shared_network_sg_account = Network.create(self.api_client,self.services["shared_network_sg"],
+ accountid=self.admin_account.name,domainid=self.admin_account.domainid,
+ networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id)
+
+ self.debug("Shared Network created: %s" % shared_network_sg_account.id)
+
+ self.debug("Deploying vm in the shared network: %s" % shared_network_sg_account.id)
+
+ vm = VirtualMachine.create(self.api_client, self.services["virtual_machine"],
+ accountid=self.admin_account.name,domainid=self.admin_account.domainid,
+ networkids=[shared_network_sg_account.id,],serviceofferingid=self.service_offering.id)
+ self.debug("Created vm %s" % vm.id)
+
+ self.debug("Trying to delete shared network: %s" % shared_network_sg_account.id)
+
+ try:
+ shared_network_sg_account.delete(self.api_client)
+ self.fail("Exception not raised while deleting network")
+ except Exception as e:
+ self.debug("Network deletion failed with exception: %s" % e)
+
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test_11_deleteSharedNwSG_DomainWide_InUse(self):
+ """ Test delete Domain wide shared network with SG which is in use"""
+
+ # Steps,
+ # 1. create a domain and its admin account
+ # 2. Create domain wide shared network with sg and subdomainaccess enabled
+ # 3. Deploy vm in the domain
+ # 4. Try to delete the shared network while its still in use
+
+ # Validations,
+ # 1. shared network deletion should fail
+
+ domain = Domain.create(self.api_client, services=self.services["domain"],
+ parentdomainid=self.domain.id)
+
+ self.debug("Created domain: %s" % domain.id)
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["acltype"] = "domain"
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ self.debug("Creating shared network domain: %s" % domain.name)
+ shared_network_sg_domain = Network.create(self.api_client,self.services["shared_network_sg"],
+ domainid=domain.id,networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id,subdomainaccess=True)
+
+ self.debug("Created shared network: %s" % shared_network_sg_domain.id)
+
+ self.debug("Deploying vm in the shared network: %s" % shared_network_sg_domain.id)
+
+ vm = VirtualMachine.create(self.api_client,self.services["virtual_machine"],domainid=domain.id,
+ networkids=[shared_network_sg_domain.id,],serviceofferingid=self.service_offering.id)
+ self.debug("Created vm %s" % vm.id)
+
+ self.debug("Trying to delete shared network: %s" % shared_network_sg_domain.id)
+
+ try:
+ shared_network_sg_domain.delete(self.api_client)
+ self.fail("Exception not raised while deleting network")
+ except Exception as e:
+ self.debug("Network deletion failed with exception: %s" % e)
+
+ self.cleanup_networks.append(shared_network_sg_domain)
+ self.cleanup_vms.append(vm)
+ self.cleanup_domains.append(domain)
+
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test_12_deleteSharedNwSGAccountSpecific_NotInUse(self):
+ """ Test delete Account specific shared network creation with SG which is not in use"""
+
+ # Steps,
+ # 1. create a user account
+ # 2. Create account specific sg enabled shared network
+ # 3. Try to delete shared network
+
+ # Validations,
+ # 1. shared network deletion should succeed
+
+ #Create admin account
+ self.admin_account = Account.create(self.api_client,self.services["account"],
+ admin=True,domainid=self.domain.id)
+
+ self.cleanup_accounts.append(self.admin_account)
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["acltype"] = "account"
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ shared_network_sg_account = Network.create(self.api_client,self.services["shared_network_sg"],
+ accountid=self.admin_account.name,domainid=self.admin_account.domainid,
+ networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id)
+
+ self.debug("Shared Network created: %s" % shared_network_sg_account.id)
+
+ try:
+ shared_network_sg_account.delete(self.api_client)
+ except Exception as e:
+ self.fail("Network deletion failed with exception: %s" % e)
+
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test_13_deleteSharedNwSG_DomainWide_NotInUse(self):
+ """ Test delete Domain wide shared network with SG which is not in use"""
+
+ # Steps,
+ # 1. create a domain and its admin account
+ # 2. Create domain wide shared network with sg and subdomainaccess enabled
+ # 4. Try to delete the shared network
+
+ # Validations,
+ # 1. shared network deletion should succeed
+
+ domain = Domain.create(self.api_client,services=self.services["domain"],
+ parentdomainid=self.domain.id)
+
+ self.debug("Created domain: %s" % domain.id)
+ self.cleanup_domains.append(domain)
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["acltype"] = "domain"
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ self.debug("Creating shared network domain: %s" % domain.name)
+ shared_network_sg_domain = Network.create(self.api_client,self.services["shared_network_sg"],
+ domainid=domain.id,networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id,subdomainaccess=True)
+
+ self.debug("Created shared network: %s" % shared_network_sg_domain.id)
+
+ self.debug("Trying to delete shared network: %s" % shared_network_sg_domain.id)
+
+ try:
+ shared_network_sg_domain.delete(self.api_client)
+ except Exception as e:
+ self.fail("Network deletion failed with exception: %s" % e)
+
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test__14_createSharedNwWithSG_withoutParams(self):
+ """ Test create shared network with SG without specifying necessary parameters"""
+
+ # Steps,
+ # 1. Test create shared network with SG without specoifying necessary parameters
+ # such as vlan, startip, endip, gateway, netmask
+ # Validations,
+ # 1. shared network creation should fail
+
+ #Create admin account
+ self.admin_account = Account.create(self.api_client,self.services["account"], admin=True,
+ domainid=self.domain.id)
+
+ self.cleanup_accounts.append(self.admin_account)
+
+ self.services["shared_network_offering_sg"]["specifyVlan"] = "True"
+ self.services["shared_network_offering_sg"]["specifyIpRanges"] = "True"
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["acltype"] = "domain"
+ self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["gateway"] = ""
+ self.services["shared_network_sg"]["startip"] = ""
+ self.services["shared_network_sg"]["endip"] = ""
+ self.services["shared_network_sg"]["netmask"] = ""
+
+
+ try:
+ self.shared_network_sg = Network.create(self.api_client,self.services["shared_network_sg"],
+ domainid=self.admin_account.domainid,
+ networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id)
+
+ self.cleanup_networks.append(self.shared_network_sg)
+ self.fail("Shared network created successfully without specifying essential parameters")
+ except Exception as e:
+ self.debug("Shared network creation failed as expected with exception: %s" % e)
+
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test__15_createVPC(self):
+ """ Test create VPC in advanced SG enabled zone"""
+
+ # Steps,
+ # 1. Try to create VPC in SG enabled advanced zone
+ # Validations,
+ # 1. VPC creation should fail
+
+ vpc_off = VpcOffering.create(self.api_client,self.services["vpc_offering"])
+ vpc_off.update(self.api_client, state='Enabled')
+
+ self.services["vpc"]["cidr"] = "10.1.1.1/16"
+ self.debug("Trying to create a VPC network in SG enabled zone: %s" %
+ self.zone.id)
+ try:
+ vpc_1 = VPC.create(self.api_client,self.services["vpc"],
+ vpcofferingid=vpc_off.id,zoneid=self.zone.id)
+ vpc_1.delete(self.api_client)
+ self.fail("VPC creation should fail in Security Group Enabled zone")
+ except Exception as e:
+ self.debug("VPC creation failed as expected with exception: %s" % e)
+ finally:
+ vpc_off.update(self.api_client, state="Disabled")
+ vpc_off.delete(self.api_client)
+
+ return
+
+@ddt
+class TestNetworksInAdvancedSG_VmOperations(cloudstackTestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ cloudstackTestClient = super(TestNetworksInAdvancedSG_VmOperations,cls).getClsTestClient()
+ cls.api_client = cloudstackTestClient.getApiClient()
+
+ cls.services = cloudstackTestClient.getConfigParser().parsedDict
+
+ # Get Zone, Domain and templates
+ cls.domain = get_domain(cls.api_client, cls.services)
+ cls.zone = get_zone(cls.api_client, cls.services)
+ cls.template = get_template(cls.api_client,cls.zone.id,cls.services["ostype"])
+
+ cls.services["virtual_machine"]["zoneid"] = cls.zone.id
+ cls.services["virtual_machine"]["template"] = cls.template.id
+
+ cls.service_offering = ServiceOffering.create(cls.api_client,cls.services["service_offering"])
+
+ cls.services["shared_network_offering_sg"]["specifyVlan"] = "True"
+ cls.services["shared_network_offering_sg"]["specifyIpRanges"] = "True"
+ #Create Network Offering
+ cls.shared_network_offering_sg = NetworkOffering.create(cls.api_client,cls.services["shared_network_offering_sg"],conservemode=False)
+
+ #Update network offering state from disabled to enabled.
+ NetworkOffering.update(cls.shared_network_offering_sg,cls.api_client,state="enabled")
+
+ cls._cleanup = [
+ cls.service_offering,
+ cls.shared_network_offering_sg
+ ]
+ return
+
+ @classmethod
+ def tearDownClass(cls):
+ try:
+ #Update network offering state from enabled to disabled.
+ cls.shared_network_offering_sg.update(cls.api_client,state="disabled")
+ #Cleanup resources used
+ cleanup_resources(cls.api_client, cls._cleanup)
+ except Exception as e:
+ raise Exception("Warning: Exception during cleanup : %s" % e)
+ return
+
+ def setUp(self):
+ self.api_client = self.testClient.getApiClient()
+ self.dbclient = self.testClient.getDbConnection()
+ self.cleanup = []
+ self.cleanup_networks = []
+ self.cleanup_accounts = []
+ self.cleanup_domains = []
+ self.cleanup_projects = []
+ self.cleanup_vms = []
+ self.cleanup_secGrps = []
+ return
+
+ def tearDown(self):
+ exceptions = []
+ try:
+ #Clean up, terminate the created network offerings
+ cleanup_resources(self.api_client, self.cleanup)
+ except Exception as e:
+ self.debug("Warning: Exception during cleanup : %s" % e)
+ exceptions.append(e)
+
+ #below components is not a part of cleanup because to mandate the order and to cleanup network
+ try:
+ for vm in self.cleanup_vms:
+ vm.delete(self.api_client)
+ except Exception as e:
+ self.debug("Warning: Exception during virtual machines cleanup : %s" % e)
+ exceptions.append(e)
+
+ # Wait for VMs to expunge
+ wait_for_cleanup(self.api_client, ["expunge.delay", "expunge.interval"])
+
+ try:
+ for sec_grp in self.cleanup_secGrps:
+ sec_grp.delete(self.api_client)
+ except Exception as e:
+ self.debug("Warning : Exception during security groups cleanup: %s" % e)
+ exceptions.append(e)
+
+ try:
+ for project in self.cleanup_projects:
+ project.delete(self.api_client)
+ except Exception as e:
+ self.debug("Warning: Exception during project cleanup : %s" % e)
+ exceptions.append(e)
+
+ try:
+ for account in self.cleanup_accounts:
+ account.delete(self.api_client)
+ except Exception as e:
+ self.debug("Warning: Exception during account cleanup : %s" % e)
+ exceptions.append(e)
+
+ #Wait till all resources created are cleaned up completely and then attempt to delete Network
+ time.sleep(self.services["sleep"])
+
+ try:
+ for network in self.cleanup_networks:
+ network.delete(self.api_client)
+ except Exception as e:
+ self.debug("Warning: Exception during network cleanup : %s" % e)
+ exceptions.append(e)
+
+ try:
+ for domain in self.cleanup_domains:
+ domain.delete(self.api_client)
+ except Exception as e:
+ self.debug("Warning: Exception during domain cleanup : %s" % e)
+ exceptions.append(e)
+
+ if len(exceptions) > 0:
+ self.faill("There were exceptions during cleanup: %s" % exceptions)
+
+ return
+
+ def setSharedNetworkParams(self, network, range=20):
+
+ # @range: range decides the endip. Pass the range as "x" if you want the difference between the startip
+ # and endip as "x"
+ # Set the subnet number of shared networks randomly prior to execution
+ # of each test case to avoid overlapping of ip addresses
+ shared_network_subnet_number = random.randrange(1,254)
+
+ self.services[network]["gateway"] = "172.16."+str(shared_network_subnet_number)+".1"
+ self.services[network]["startip"] = "172.16."+str(shared_network_subnet_number)+".2"
+ self.services[network]["endip"] = "172.16."+str(shared_network_subnet_number)+"."+str(range+1)
+ self.services[network]["netmask"] = "255.255.255.0"
+
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test__16_AccountSpecificNwAccess(self):
+ """ Test account specific network access of users"""
+
+ # Steps,
+ # 1. create multiple accounts/users and their account specific SG enabled shared networks
+ # 2. Deploy VM in the account specific network with user of that account
+ # 3. Try to deploy VM in one account specific network from other account user
+
+ # Validations,
+ # 1. VM deployment should be allowed for the users of the same account only for their account
+ # specific networks respectively
+
+ self.debug("Creating user account 1")
+ account_1 = Account.create(self.api_client,self.services["account"],
+ domainid=self.domain.id)
+ self.debug("Created account : %s" % account_1.name)
+
+ self.cleanup_accounts.append(account_1)
+
+ self.debug("Creating user account 2")
+ account_2 = Account.create(self.api_client,self.services["account"],
+ domainid=self.domain.id)
+ self.debug("Created account: %s" % account_2.name)
+
+ self.cleanup_accounts.append(account_2)
+
+ self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
+ self.services["shared_network_sg"]["acltype"] = "account"
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ shared_network_account_1 = Network.create(self.api_client,self.services["shared_network_sg"],
+ accountid=account_1.name,domainid=account_1.domainid,
+ networkofferingid=self.shared_network_offering_sg.id,zoneid=self.zone.id)
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ shared_network_account_2 = Network.create(self.api_client,self.services["shared_network_sg"],
+ accountid=account_2.name,domainid=account_2.domainid,
+ networkofferingid=self.shared_network_offering_sg.id,
+ zoneid=self.zone.id)
+
+ # Creaint user API client of account 1
+ api_client_account_1 = self.testClient.createUserApiClient(UserName=account_1.name,
+ DomainName=account_1.domain)
+
+ # Creaint user API client of account 2
+ api_client_account_2 = self.testClient.createUserApiClient(UserName=account_2.name,
+ DomainName=account_2.domain)
+
+ self.debug("Creating virtual machine in account %s with account specific shared network %s" %
+ (account_1.name, shared_network_account_1.id))
+
+ VirtualMachine.create(api_client_account_1,self.services["virtual_machine"],
+ templateid=self.template.id,accountid=account_1.name,
+ domainid=account_1.domainid,networkids=[shared_network_account_1.id,],
+ serviceofferingid=self.service_offering.id)
+
+ self.debug("Creating virtual machine in account %s with account specific shared network %s" %
+ (account_2.name, shared_network_account_2.id))
+
+ VirtualMachine.create(api_client_account_2,self.services["virtual_machine"],
+ templateid=self.template.id,accountid=account_2.name,
+ domainid=account_2.domainid,networkids=[shared_network_account_2.id,],
+ serviceofferingid=self.service_offering.id)
+
+ try:
+ self.debug("Trying to create virtual machine in account specific shared network of\
+ account %s using the api client of account %s, this should fail" %
+ (account_2.name, account_1.name))
+
+ VirtualMachine.create(api_client_account_2,self.services["virtual_machine"],
+ templateid=self.template.id,accountid=account_1.name,
+ domainid=account_1.domainid,networkids=[shared_network_account_1.id,],
+ serviceofferingid=self.service_offering.id)
+
+ self.fail("Vm creation succeded, should have failed")
+ except Exception as e:
+ self.debug("VM creation failed as expected with exception: %s" % e)
+
+ return
+
+ @attr(tags = ["advancedsg"])
+ def test__17_DomainSpecificNwAccess(self):
+ """ Test domain specific network access of users"""
+
+ # Steps,
+ # 1. create multiple domains with accounts/users in them and their domain wide SG enabled shared networks
+ # 2. Deploy VMs in the domain wide network with user belonging to that domain
+ # 3. Try to deploy VM in domain wide network with user of other domain
+
+ # Validations,
+ # 1. VM deployment should be allowed only for the users of the same domain for their domain
+ # wide networks respectively
+
+ self.debug("Creating domain 1")
+ domain_1 = Domain.create(self.api_client,services=self.services["domain"],
+ parentdomainid=self.domain.id)
+
+ self.debug("Created domain %s" % domain_1.name)
+ self.cleanup_domains.append(domain_1)
+
+ self.debug("Creating domain 2")
+ domain_2 = Domain.create(self.api_client,services=self.services["domain"],
+ parentdomainid=self.domain.id)
+
+ self.debug("Created domain: %s" % domain_2.name)
+ self.cleanup_domains.append(domain_2)
+
+ self.debug("Creating user account under domain %s" % domain_1.name)
+ account_1 = Account.create(self.api_client,self.services["account"],
+ domainid=domain_1.id)
+ self.debug("Created account : %s" % account_1.name)
+ self.cleanup_accounts.append(account_1)
+ self.debug("Creating user account under domain %s" % domain_2.name)
+ account_2 = Account.create(self.api_client,self.services["account"],
+ domainid=domain_2.id)
+
+ self.debug("Created account: %s" % account_2.name)
+ self.cleanup_accounts.append(account_2)
+
+ self.services["shared_network_sg"]["networkofferingid"] = self.shared_network_offering_sg.id
+ self.services["shared_network_sg"]["acltype"] = "domain"
+
+ physical_network, vlan = get_free_vlan(self.api_client, self.zone.id)
+
+ #create network using the shared network offering created
+ self.services["shared_network_sg"]["vlan"] = vlan
+ self.services["shared_network_sg"]["physicalnetworkid"] = physical_network.id
+
+ self.setSharedNetworkParams("shared_network_sg")
+
+ shared_network_domain_1 = Network.create(self.api_client,self.services["shared_network_sg"],
<TRUNCATED>
[3/3] git commit: updated refs/heads/4.3 to 029f4e3
Posted by gi...@apache.org.
CLOUDSTACK-2237: Automation - Adding new test cases for security group in advanced zone feature
Project: http://git-wip-us.apache.org/repos/asf/cloudstack/repo
Commit: http://git-wip-us.apache.org/repos/asf/cloudstack/commit/029f4e36
Tree: http://git-wip-us.apache.org/repos/asf/cloudstack/tree/029f4e36
Diff: http://git-wip-us.apache.org/repos/asf/cloudstack/diff/029f4e36
Branch: refs/heads/4.3
Commit: 029f4e36a0188520f46d7997df3e53ab1661fb63
Parents: f908a15
Author: Ashutosh K <as...@clogeny.com>
Authored: Fri Dec 27 15:55:41 2013 +0530
Committer: Girish Shilamkar <gi...@clogeny.com>
Committed: Fri Dec 27 16:01:33 2013 +0530
----------------------------------------------------------------------
.../component/test_advancedsg_networks.py | 2420 +++++++++++++-----
tools/marvin/marvin/config/config.cfg | 159 ++
tools/marvin/marvin/configGenerator.py | 5 +-
tools/marvin/marvin/integration/lib/base.py | 2 +
tools/marvin/marvin/integration/lib/common.py | 50 +-
5 files changed, 1997 insertions(+), 639 deletions(-)
----------------------------------------------------------------------