mirror of
https://github.com/apache/cloudstack
synced 2026-08-02 05:26:35 +00:00
188 lines
7.1 KiB
Python
188 lines
7.1 KiB
Python
# 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.
|
|
|
|
"""Smoke tests for firewall rules on VPC public IPs."""
|
|
|
|
from nose.plugins.attrib import attr
|
|
|
|
from marvin.cloudstackTestCase import cloudstackTestCase
|
|
from marvin.lib.base import Account, FireWallRule, Network, NetworkOffering, PublicIPAddress, VPC, VpcOffering
|
|
from marvin.lib.common import get_domain, get_zone, list_publicIP
|
|
from marvin.lib.utils import cleanup_resources, wait_until
|
|
|
|
|
|
class TestVpcFirewallRules(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(TestVpcFirewallRules, cls).getClsTestClient()
|
|
cls.apiclient = cls.testClient.getApiClient()
|
|
cls.services = cls.testClient.getParsedTestDataConfig()
|
|
cls.zone = get_zone(cls.apiclient, cls.testClient.getZoneForTests())
|
|
cls.domain = get_domain(cls.apiclient)
|
|
cls._cleanup = []
|
|
|
|
cls.account = Account.create(
|
|
cls.apiclient,
|
|
cls.services["account"],
|
|
domainid=cls.domain.id
|
|
)
|
|
cls._cleanup.append(cls.account)
|
|
|
|
cls.services["vpc_offering"]["supportedservices"] = (
|
|
"Vpn,Dhcp,Dns,SourceNat,Lb,UserData,StaticNat,"
|
|
"NetworkACL,PortForwarding,Firewall"
|
|
)
|
|
cls.services["vpc_offering"]["serviceProviderList"] = {
|
|
"Vpn": "VpcVirtualRouter",
|
|
"Dhcp": "VpcVirtualRouter",
|
|
"Dns": "VpcVirtualRouter",
|
|
"SourceNat": "VpcVirtualRouter",
|
|
"Lb": "VpcVirtualRouter",
|
|
"UserData": "VpcVirtualRouter",
|
|
"StaticNat": "VpcVirtualRouter",
|
|
"NetworkACL": "VpcVirtualRouter",
|
|
"PortForwarding": "VpcVirtualRouter",
|
|
"Firewall": "VpcVirtualRouter"
|
|
}
|
|
|
|
cls.vpc_offering = VpcOffering.create(
|
|
cls.apiclient,
|
|
cls.services["vpc_offering"]
|
|
)
|
|
cls.vpc_offering.update(cls.apiclient, state="Enabled")
|
|
cls._cleanup.append(cls.vpc_offering)
|
|
|
|
network_offering = NetworkOffering.list(
|
|
cls.apiclient,
|
|
name="DefaultIsolatedNetworkOfferingForVpcNetworks"
|
|
)
|
|
cls.assertTrue(
|
|
network_offering is not None and len(network_offering) > 0,
|
|
"No VPC tier network offering found"
|
|
)
|
|
cls.network_offering = network_offering[0]
|
|
cls.services["vpc"]["cidr"] = "10.20.30.0/24"
|
|
cls.vpc = VPC.create(
|
|
cls.apiclient,
|
|
cls.services["vpc"],
|
|
vpcofferingid=cls.vpc_offering.id,
|
|
zoneid=cls.zone.id,
|
|
account=cls.account.name,
|
|
domainid=cls.account.domainid
|
|
)
|
|
|
|
cls.tier = Network.create(
|
|
cls.apiclient,
|
|
services={
|
|
"name": "vpc-fw-tier",
|
|
"displaytext": "vpc-fw-tier"
|
|
},
|
|
accountid=cls.account.name,
|
|
domainid=cls.account.domainid,
|
|
networkofferingid=cls.network_offering.id,
|
|
zoneid=cls.zone.id,
|
|
vpcid=cls.vpc.id,
|
|
gateway="10.20.30.1",
|
|
netmask="255.255.255.0"
|
|
)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
try:
|
|
cleanup_resources(cls.apiclient, cls._cleanup)
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup: %s" % e)
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.cleanup = []
|
|
|
|
def tearDown(self):
|
|
cleanup_resources(self.apiclient, self.cleanup)
|
|
|
|
def _wait_for_firewall_rule(self, rule_id):
|
|
rules = FireWallRule.list(self.apiclient, id=rule_id, listall=True)
|
|
if rules and len(rules) == 1:
|
|
return True, rules[0]
|
|
return False, None
|
|
|
|
@attr(tags=["advanced", "advancedns", "smoke"], required_hardware="false")
|
|
def test_01_create_firewall_rule_on_vpc_public_ip(self):
|
|
"""Verify firewall rule can be created and listed on a dedicated VPC public IP."""
|
|
public_ip = PublicIPAddress.create(
|
|
self.apiclient,
|
|
zoneid=self.zone.id,
|
|
accountid=self.account.name,
|
|
domainid=self.account.domainid,
|
|
vpcid=self.vpc.id
|
|
)
|
|
self.cleanup.append(public_ip)
|
|
|
|
firewall_rule = FireWallRule.create(
|
|
self.apiclient,
|
|
ipaddressid=public_ip.ipaddress.id,
|
|
protocol="tcp",
|
|
cidrlist=["0.0.0.0/0"],
|
|
startport=19090,
|
|
endport=19090,
|
|
vpcid=self.vpc.id
|
|
)
|
|
self.cleanup.insert(0, firewall_rule)
|
|
|
|
result, listed_rule = wait_until(2, 10, self._wait_for_firewall_rule, firewall_rule.id)
|
|
self.assertTrue(result, "Firewall rule was not listed for the VPC public IP")
|
|
self.assertEqual(listed_rule.id, firewall_rule.id)
|
|
self.assertEqual(listed_rule.ipaddressid, public_ip.ipaddress.id)
|
|
self.assertEqual(listed_rule.vpcid, self.vpc.id)
|
|
self.assertEqual(listed_rule.protocol.lower(), "tcp")
|
|
self.assertEqual(int(listed_rule.startport), 19090)
|
|
self.assertEqual(int(listed_rule.endport), 19090)
|
|
|
|
@attr(tags=["advanced", "advancedns", "smoke"], required_hardware="false")
|
|
def test_02_create_firewall_rule_on_vpc_source_nat_ip(self):
|
|
"""Verify firewall rule can be created and listed on the VPC source NAT IP."""
|
|
source_nat_ips = list_publicIP(
|
|
self.apiclient,
|
|
vpcid=self.vpc.id,
|
|
listall=True,
|
|
issourcenat=True
|
|
)
|
|
self.assertTrue(source_nat_ips is not None and len(source_nat_ips) > 0,
|
|
"No source NAT IP found for the VPC")
|
|
source_nat_ip = source_nat_ips[0]
|
|
|
|
firewall_rule = FireWallRule.create(
|
|
self.apiclient,
|
|
ipaddressid=source_nat_ip.id,
|
|
protocol="tcp",
|
|
cidrlist=["0.0.0.0/0"],
|
|
startport=19443,
|
|
endport=19443,
|
|
vpcid=self.vpc.id
|
|
)
|
|
self.cleanup.append(firewall_rule)
|
|
|
|
result, listed_rule = wait_until(2, 10, self._wait_for_firewall_rule, firewall_rule.id)
|
|
self.assertTrue(result, "Firewall rule was not listed for the VPC source NAT IP")
|
|
self.assertEqual(listed_rule.id, firewall_rule.id)
|
|
self.assertEqual(listed_rule.ipaddressid, source_nat_ip.id)
|
|
self.assertEqual(listed_rule.vpcid, self.vpc.id)
|
|
self.assertEqual(listed_rule.protocol.lower(), "tcp")
|
|
self.assertEqual(int(listed_rule.startport), 19443)
|
|
self.assertEqual(int(listed_rule.endport), 19443)
|
