mirror of
https://github.com/apache/cloudstack.git
synced 2025-11-02 11:52:28 +01:00
CLOUDSTACK-8971 - Improve the current test_privategw_acls.py
- Read cofiguration from the zone: template id; template id; and domain id - Create Account and VPC Offering - Add cleanup resources
This commit is contained in:
Wilder Rodrigues
parent
6fe5ae0d60
commit
8708d0dc43
@ -24,88 +24,239 @@ from marvin.lib.base import *
|
||||
from marvin.lib.common import *
|
||||
from nose.plugins.attrib import attr
|
||||
|
||||
import logging
|
||||
|
||||
class Services:
|
||||
"""Test VPC network services - Port Forwarding Rules Test Data Class.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.services = {
|
||||
"account": {
|
||||
"email": "test@test.com",
|
||||
"firstname": "Test",
|
||||
"lastname": "User",
|
||||
"username": "test",
|
||||
# Random characters are appended for unique
|
||||
# username
|
||||
"password": "password",
|
||||
},
|
||||
"host1": None,
|
||||
"service_offering": {
|
||||
"name": "Tiny Instance",
|
||||
"displaytext": "Tiny Instance",
|
||||
"cpunumber": 1,
|
||||
"cpuspeed": 100,
|
||||
"memory": 128,
|
||||
},
|
||||
"network_offering": {
|
||||
"name": 'VPC Network offering',
|
||||
"displaytext": 'VPC Network off',
|
||||
"guestiptype": 'Isolated',
|
||||
"supportedservices": 'Vpn,Dhcp,Dns,SourceNat,PortForwarding,Lb,UserData,StaticNat,NetworkACL',
|
||||
"traffictype": 'GUEST',
|
||||
"availability": 'Optional',
|
||||
"useVpc": 'on',
|
||||
"serviceProviderList": {
|
||||
"Vpn": 'VpcVirtualRouter',
|
||||
"Dhcp": 'VpcVirtualRouter',
|
||||
"Dns": 'VpcVirtualRouter',
|
||||
"SourceNat": 'VpcVirtualRouter',
|
||||
"PortForwarding": 'VpcVirtualRouter',
|
||||
"Lb": 'VpcVirtualRouter',
|
||||
"UserData": 'VpcVirtualRouter',
|
||||
"StaticNat": 'VpcVirtualRouter',
|
||||
"NetworkACL": 'VpcVirtualRouter'
|
||||
},
|
||||
},
|
||||
"network_offering_no_lb": {
|
||||
"name": 'VPC Network offering',
|
||||
"displaytext": 'VPC Network off',
|
||||
"guestiptype": 'Isolated',
|
||||
"supportedservices": 'Dhcp,Dns,SourceNat,PortForwarding,UserData,StaticNat,NetworkACL',
|
||||
"traffictype": 'GUEST',
|
||||
"availability": 'Optional',
|
||||
"useVpc": 'on',
|
||||
"serviceProviderList": {
|
||||
"Dhcp": 'VpcVirtualRouter',
|
||||
"Dns": 'VpcVirtualRouter',
|
||||
"SourceNat": 'VpcVirtualRouter',
|
||||
"PortForwarding": 'VpcVirtualRouter',
|
||||
"UserData": 'VpcVirtualRouter',
|
||||
"StaticNat": 'VpcVirtualRouter',
|
||||
"NetworkACL": 'VpcVirtualRouter'
|
||||
},
|
||||
},
|
||||
"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'
|
||||
},
|
||||
"network": {
|
||||
"name": "Test Network",
|
||||
"displaytext": "Test Network",
|
||||
"netmask": '255.255.255.0'
|
||||
},
|
||||
"virtual_machine": {
|
||||
"displayname": "Test VM",
|
||||
"username": "root",
|
||||
"password": "password",
|
||||
"ssh_port": 22,
|
||||
"privateport": 22,
|
||||
"publicport": 22,
|
||||
"protocol": 'TCP',
|
||||
},
|
||||
"ostype": 'CentOS 5.3 (64-bit)',
|
||||
"timeout": 10,
|
||||
}
|
||||
|
||||
class TestPrivateGwACL(cloudstackTestCase):
|
||||
def setUp(self):
|
||||
self.apiClient = self.testClient.getApiClient()
|
||||
self.networkOfferingId = 11
|
||||
self.networkId = None
|
||||
self.vmId = None
|
||||
self.vpcId = None
|
||||
self.aclId = None
|
||||
self.zoneId = 1
|
||||
self.serviceOfferingId = 1
|
||||
self.templateId = 5
|
||||
self.privateGwId = None
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
|
||||
cls.testClient = super(TestPrivateGwACL, cls).getClsTestClient()
|
||||
cls.api_client = cls.testClient.getApiClient()
|
||||
|
||||
cls.services = Services().services
|
||||
# Get Zone, Domain and templates
|
||||
cls.domain = get_domain(cls.api_client)
|
||||
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
||||
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._cleanup = [cls.service_offering]
|
||||
|
||||
cls.logger = logging.getLogger('TestPrivateGwACL')
|
||||
cls.stream_handler = logging.StreamHandler()
|
||||
cls.logger.setLevel(logging.DEBUG)
|
||||
cls.logger.addHandler(cls.stream_handler)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
try:
|
||||
cleanup_resources(cls.api_client, cls._cleanup)
|
||||
except Exception as e:
|
||||
raise Exception("Warning: Exception during cleanup : %s" % e)
|
||||
return
|
||||
|
||||
def setUp(self):
|
||||
self.apiclient = self.testClient.getApiClient()
|
||||
|
||||
self.logger.debug("Creating Admin Account for Domain ID ==> %s" %self.domain.id)
|
||||
self.account = Account.create(
|
||||
self.apiclient,
|
||||
self.services["account"],
|
||||
admin=True,
|
||||
domainid=self.domain.id)
|
||||
|
||||
self.logger.debug("Creating a VPC offering..")
|
||||
self.vpc_off = VpcOffering.create(
|
||||
self.apiclient,
|
||||
self.services["vpc_offering"])
|
||||
|
||||
self.logger.debug("Enabling the VPC offering created")
|
||||
self.vpc_off.update(self.apiclient, state='Enabled')
|
||||
|
||||
self.logger.debug("Creating a VPC network in the account: %s" % self.account.name)
|
||||
self.services["vpc"]["cidr"] = '10.1.1.1/16'
|
||||
self.vpc = VPC.create(
|
||||
self.apiclient,
|
||||
self.services["vpc"],
|
||||
vpcofferingid=self.vpc_off.id,
|
||||
zoneid=self.zone.id,
|
||||
account=self.account.name,
|
||||
domainid=self.account.domainid)
|
||||
|
||||
self.cleanup = [self.vpc, self.vpc_off, self.account]
|
||||
return
|
||||
|
||||
def tearDown(self):
|
||||
try:
|
||||
cleanup_resources(self.apiclient, self.cleanup)
|
||||
except Exception as e:
|
||||
raise Exception("Warning: Exception during cleanup : %s" % e)
|
||||
return
|
||||
|
||||
@attr(tags=["advanced"], required_hardware="false")
|
||||
def test_privategw_acl(self):
|
||||
#TODO: SIMENH: add a new test to verification of ACL rules
|
||||
|
||||
# 1) Create VPC
|
||||
self.createVPC()
|
||||
|
||||
# 2) Create ACl
|
||||
self.createACL()
|
||||
|
||||
# 3) Create ACl Item
|
||||
self.createACLItem()
|
||||
|
||||
# 4) Create network with ACL
|
||||
self.createNetwork()
|
||||
|
||||
# 5) create private gw
|
||||
self.createPvtGw()
|
||||
|
||||
# 6) update acl id
|
||||
self.replaceacl()
|
||||
|
||||
def createACL(self):
|
||||
createAclCmd = createNetworkACLList.createNetworkACLListCmd()
|
||||
createAclCmd.name = "acl1"
|
||||
createAclCmd.description = "new acl"
|
||||
createAclCmd.vpcid = self.vpcId
|
||||
createAclResponse = self.apiClient.createNetworkACLList(createAclCmd)
|
||||
createAclCmd.vpcid = self.vpc.id
|
||||
createAclResponse = self.apiclient.createNetworkACLList(createAclCmd)
|
||||
|
||||
self.aclId = createAclResponse.id
|
||||
|
||||
self.assertIsNotNone(self.aclId, "Failed to create ACL.")
|
||||
|
||||
def createACLItem(self):
|
||||
createAclItemCmd = createNetworkACL.createNetworkACLCmd()
|
||||
createAclItemCmd.cidr = "0.0.0.0/0"
|
||||
createAclItemCmd.protocol = "TCP"
|
||||
createAclItemCmd.number = "10"
|
||||
createAclItemCmd.number = "1"
|
||||
createAclItemCmd.action = "Deny"
|
||||
createAclItemCmd.aclid = self.aclId
|
||||
createAclItemResponse = self.apiClient.createNetworkACL(createAclItemCmd)
|
||||
self.assertIsNotNone(createAclItemResponse.id, "Network failed to aclItem")
|
||||
|
||||
def createVPC(self):
|
||||
createVPCCmd = createVPC.createVPCCmd()
|
||||
createVPCCmd.name = "new vpc"
|
||||
createVPCCmd.cidr = "10.1.1.0/24"
|
||||
createVPCCmd.displaytext = "new vpc"
|
||||
createVPCCmd.vpcofferingid = 1
|
||||
createVPCCmd.zoneid = self.zoneId
|
||||
createVPCResponse = self.apiClient.createVPC(createVPCCmd)
|
||||
self.vpcId = createVPCResponse.id
|
||||
createAclItemResponse = self.apiclient.createNetworkACL(createAclItemCmd)
|
||||
|
||||
self.assertIsNotNone(createAclItemResponse.id, "Failed to create ACL item.")
|
||||
|
||||
def createNetwork(self):
|
||||
createNetworkCmd = createNetwork.createNetworkCmd()
|
||||
createNetworkCmd.name = "vpc network"
|
||||
createNetworkCmd.displaytext = "vpc network"
|
||||
createNetworkCmd.netmask = "255.255.255.0"
|
||||
createNetworkCmd.gateway = "10.1.1.1"
|
||||
createNetworkCmd.zoneid = self.zoneId
|
||||
createNetworkCmd.vpcid = self.vpcId
|
||||
createNetworkCmd.networkofferingid = self.networkOfferingId
|
||||
createNetworkCmd.aclid = self.aclId
|
||||
createNetworkResponse = self.apiClient.createNetwork(createNetworkCmd)
|
||||
try:
|
||||
self.logger.debug('Create NetworkOffering')
|
||||
net_offerring = self.services["network_offering"]
|
||||
net_offerring["name"] = "NET_OFF-10.1.1.1"
|
||||
nw_off = NetworkOffering.create(
|
||||
self.apiclient,
|
||||
net_offerring,
|
||||
conservemode=False)
|
||||
|
||||
self.assertIsNotNone(createNetworkResponse.id, "Network failed to create")
|
||||
self.networkId = createNetworkResponse.id
|
||||
nw_off.update(self.apiclient, state='Enabled')
|
||||
|
||||
self.logger.debug('Created and Enabled NetworkOffering')
|
||||
|
||||
self.services["network"]["name"] = "NETWORK-10.1.1.1"
|
||||
|
||||
self.logger.debug('Adding Network=%s' % self.services["network"])
|
||||
obj_network = Network.create(
|
||||
self.apiclient,
|
||||
self.services["network"],
|
||||
accountid=self.account.name,
|
||||
domainid=self.account.domainid,
|
||||
networkofferingid=nw_off.id,
|
||||
zoneid=self.zone.id,
|
||||
gateway="10.1.1.1",
|
||||
vpcid=self.vpc.id
|
||||
)
|
||||
|
||||
self.logger.debug("Created network with ID: %s" % obj_network.id)
|
||||
except Exception, e:
|
||||
self.fail('Unable to create a Network with offering=%s because of %s ' % (net_offerring, e))
|
||||
|
||||
self.network = obj_network
|
||||
|
||||
self.cleanup.insert(0, nw_off)
|
||||
self.cleanup.insert(0, obj_network)
|
||||
|
||||
def createPvtGw(self):
|
||||
createPrivateGatewayCmd = createPrivateGateway.createPrivateGatewayCmd()
|
||||
@ -114,18 +265,23 @@ class TestPrivateGwACL(cloudstackTestCase):
|
||||
createPrivateGatewayCmd.netmask = "255.255.255.0"
|
||||
createPrivateGatewayCmd.ipaddress = "10.147.30.200"
|
||||
createPrivateGatewayCmd.vlan = "30"
|
||||
createPrivateGatewayCmd.vpcid = self.vpcId
|
||||
createPrivateGatewayCmd.vpcid = self.vpc.id
|
||||
createPrivateGatewayCmd.sourcenatsupported = "true"
|
||||
createPrivateGatewayCmd.aclid = self.aclId
|
||||
privateGatewayResponse = self.apiClient.createPrivateGateway(createPrivateGatewayCmd)
|
||||
|
||||
try:
|
||||
privateGatewayResponse = self.apiclient.createPrivateGateway(createPrivateGatewayCmd)
|
||||
except Exception as e:
|
||||
self.fail("Failed to create Private Gateway ==> %s" % e)
|
||||
|
||||
self.privateGwId = privateGatewayResponse.id
|
||||
|
||||
self.assertIsNotNone(self.privateGwId, "Failed to create ACL.")
|
||||
|
||||
def replaceacl(self):
|
||||
replaceNetworkACLListCmd = replaceNetworkACLList.replaceNetworkACLListCmd()
|
||||
replaceNetworkACLListCmd.aclid = self.aclId
|
||||
replaceNetworkACLListCmd.gatewayid = self.privateGwId
|
||||
successResponse = self.apiClient.replaceNetworkACLList(replaceNetworkACLListCmd);
|
||||
successResponse = self.apiclient.replaceNetworkACLList(replaceNetworkACLListCmd);
|
||||
|
||||
def tearDown(self):
|
||||
#destroy the vm
|
||||
return
|
||||
self.assertTrue(successResponse.success, "Failed to replace ACL list.")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user