CLOUDSTACK-8971 - Improve the current test_privategw_acls.py

test/integration/smoke/test_privategw_acl.py

@@ -24,88 +24,239 @@
 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):
+
+    @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.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
+        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)
 
-    @attr(tags=["advanced"], required_hardware="false")
-    def test_privategw_acl(self):
-        #TODO: SIMENH: add a new test to verification of ACL rules
+        self.logger.debug("Creating a VPC offering..")
+        self.vpc_off = VpcOffering.create(
+            self.apiclient,
+            self.services["vpc_offering"])
 
-        # 1) Create VPC
-        self.createVPC()
+        self.logger.debug("Enabling the VPC offering created")
+        self.vpc_off.update(self.apiclient, state='Enabled')
 
-        # 2) Create ACl
-        self.createACL()
+        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
 
-        # 3) Create ACl Item
-        self.createACLItem()
+    def tearDown(self):
+        try:
+            cleanup_resources(self.apiclient, self.cleanup)
+        except Exception as e:
+            raise Exception("Warning: Exception during cleanup : %s" % e)
+        return
 
-        # 4) Create network with ACL
+    @attr(tags=["advanced"], required_hardware="false")
+    def test_privategw_acl(self):
+        self.createACL()
+        self.createACLItem()
         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)
+
+            nw_off.update(self.apiclient, state='Enabled')
 
-        self.assertIsNotNone(createNetworkResponse.id, "Network failed to create")
-        self.networkId = createNetworkResponse.id
+            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 @@ def createPvtGw(self):
         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.")