mirror of
https://github.com/apache/cloudstack.git
synced 2025-10-26 08:42:29 +01:00
897 lines
34 KiB
Python
897 lines
34 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.
|
|
""" P1 tests for VPN users
|
|
"""
|
|
# Import Local Modules
|
|
from nose.plugins.attrib import attr
|
|
from marvin.cloudstackTestCase import cloudstackTestCase
|
|
from marvin.lib.base import (
|
|
Account,
|
|
ServiceOffering,
|
|
VirtualMachine,
|
|
PublicIPAddress,
|
|
Network,
|
|
LoadBalancerRule,
|
|
Alert,
|
|
Router,
|
|
Vpn,
|
|
FireWallRule
|
|
)
|
|
from marvin.lib.common import (get_domain,
|
|
get_zone,
|
|
get_template
|
|
)
|
|
from marvin.lib.utils import (cleanup_resources,
|
|
random_gen,
|
|
verifyRouterState)
|
|
from marvin.cloudstackAPI import createLBStickinessPolicy
|
|
from marvin.codes import PASS
|
|
from marvin.sshClient import SshClient
|
|
|
|
|
|
class TestHAProxyStickyness(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(TestHAProxyStickyness, cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
|
|
cls.services = cls.testClient.getParsedTestDataConfig()
|
|
# 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["service_offering"]["name"] = "Medium Instance"
|
|
cls.services["service_offering"]["cpuspeed"] = "1024"
|
|
cls.services["service_offering"]["memory"] = "1024"
|
|
cls.service_offering = ServiceOffering.create(
|
|
cls.api_client,
|
|
cls.services["service_offering"]
|
|
)
|
|
cls._cleanup = [cls.service_offering, ]
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
try:
|
|
# Cleanup resources used
|
|
cleanup_resources(cls.api_client, cls._cleanup)
|
|
pass
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
return
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.account = Account.create(
|
|
self.apiclient,
|
|
self.services["account"],
|
|
domainid=self.domain.id
|
|
)
|
|
self.virtual_machine = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid=self.account.name,
|
|
domainid=self.account.domainid,
|
|
serviceofferingid=self.service_offering.id
|
|
)
|
|
|
|
self.virtual_machine_2 = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid=self.account.name,
|
|
domainid=self.account.domainid,
|
|
serviceofferingid=self.service_offering.id
|
|
)
|
|
self.public_ip = PublicIPAddress.create(
|
|
self.apiclient,
|
|
self.virtual_machine.account,
|
|
self.virtual_machine.zoneid,
|
|
self.virtual_machine.domainid,
|
|
self.services["virtual_machine"]
|
|
)
|
|
FireWallRule.create(
|
|
self.apiclient,
|
|
ipaddressid=self.public_ip.ipaddress.id,
|
|
protocol='TCP',
|
|
cidrlist=[self.services["fwrule"]["cidr"]],
|
|
startport=self.services["fwrule"]["startport"],
|
|
endport=self.services["fwrule"]["endport"]
|
|
)
|
|
self.cleanup = [self.account, ]
|
|
return
|
|
|
|
def tearDown(self):
|
|
try:
|
|
# Clean up, terminate the created instance, volumes and snapshots
|
|
cleanup_resources(self.apiclient, self.cleanup)
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
return
|
|
|
|
def get_Network(self, account):
|
|
"""Returns a network for account"""
|
|
|
|
networks = Network.list(
|
|
self.apiclient,
|
|
account=account.name,
|
|
domainid=account.domainid,
|
|
listall=True
|
|
)
|
|
self.assertIsInstance(networks,
|
|
list,
|
|
"List networks should return a valid response")
|
|
return networks[0]
|
|
|
|
def create_LB_Rule(self, public_ip, network, vmarray, services=None):
|
|
"""Create and validate the load balancing rule"""
|
|
|
|
self.debug("Creating LB rule for IP address: %s" %
|
|
public_ip.ipaddress.ipaddress)
|
|
objservices = None
|
|
if services:
|
|
objservices = services
|
|
else:
|
|
objservices = self.services["lbrule"]
|
|
|
|
self.services["lbrule"]["publicport"] = 22
|
|
self.services["lbrule"]["privateport"] = 22
|
|
|
|
lb_rule = LoadBalancerRule.create(
|
|
self.apiclient,
|
|
objservices,
|
|
ipaddressid=public_ip.ipaddress.id,
|
|
accountid=self.account.name,
|
|
networkid=network.id,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.debug("Adding virtual machines %s to LB rule" % str(vmarray))
|
|
lb_rule.assign(self.apiclient, vmarray)
|
|
return lb_rule
|
|
|
|
def configure_Stickiness_Policy(self, lb_rule, method, paramDict=None):
|
|
"""Configure the stickiness policy on lb rule"""
|
|
try:
|
|
result = lb_rule.createSticky(
|
|
self.apiclient,
|
|
methodname=method,
|
|
name="-".join([method, random_gen()]),
|
|
param=paramDict
|
|
)
|
|
self.debug("Response: %s" % result)
|
|
return result
|
|
except Exception as e:
|
|
self.fail("Configure sticky policy failed with exception: %s" % e)
|
|
|
|
def validate_Stickiness_Policy(self, lb_rule, method, publicip):
|
|
"""Validates the stickiness policy"""
|
|
|
|
sticky_policies = lb_rule.listStickyPolicies(self.apiclient,
|
|
lbruleid=lb_rule.id,
|
|
listall=True)
|
|
self.assertIsInstance(
|
|
sticky_policies,
|
|
list,
|
|
"List sticky policies should return a valid list")
|
|
sticky_policy = sticky_policies[0]
|
|
|
|
self.debug("Stickiness policy method: %s" %
|
|
sticky_policy.stickinesspolicy[0].methodname)
|
|
self.assertEqual(
|
|
sticky_policy.stickinesspolicy[0].methodname,
|
|
method,
|
|
"Stickiness policy should have method as - %s" %
|
|
method)
|
|
|
|
hostnames = []
|
|
|
|
hostnames = self.try_ssh(publicip, hostnames)
|
|
hostnames = self.try_ssh(publicip, hostnames)
|
|
|
|
self.debug("hostnames: %s" % hostnames)
|
|
self.debug("set(hostnames): %s" % set(hostnames))
|
|
|
|
# For each ssh, host should be the same, else stickiness policy is not
|
|
# working properly
|
|
if len(hostnames) == len(set(hostnames)):
|
|
raise Exception(
|
|
"Stickyness policy: %s not working properly, got hostnames %s" %
|
|
(method, hostnames))
|
|
return
|
|
|
|
def delete_Stickiness_policy(self, policy, lb_rule):
|
|
"""Deletes the stickiness policy"""
|
|
|
|
try:
|
|
lb_rule.deleteSticky(self.apiclient, id=policy.id)
|
|
except Exception as e:
|
|
self.fail("Failed to delete the stickiness policy: %s" % e)
|
|
|
|
sticky_policies = lb_rule.listStickyPolicies(self.apiclient,
|
|
lbruleid=lb_rule.id,
|
|
listall=True)
|
|
self.assertIsInstance(
|
|
sticky_policies,
|
|
list,
|
|
"List stickiness policies shall return a valid response")
|
|
|
|
policy = sticky_policies[0]
|
|
|
|
self.assertEqual(len(policy.stickinesspolicy),
|
|
0,
|
|
"List stickiness policy should return nothing")
|
|
return
|
|
|
|
def check_stickiness_supported_methods(self, supportedMethods, value):
|
|
|
|
for i, dic in enumerate(supportedMethods):
|
|
if dic["methodname"] == value:
|
|
return True
|
|
return False
|
|
|
|
def acquire_Public_Ip(self):
|
|
"""Acquires the public IP"""
|
|
|
|
try:
|
|
self.debug("Acquiring public IP for account: %s" %
|
|
self.account.name)
|
|
public_ip = PublicIPAddress.create(
|
|
self.apiclient,
|
|
self.virtual_machine.account,
|
|
self.virtual_machine.zoneid,
|
|
self.virtual_machine.domainid,
|
|
self.services["virtual_machine"]
|
|
)
|
|
self.debug("Acquired public IP: %s" %
|
|
public_ip.ipaddress.ipaddress)
|
|
|
|
FireWallRule.create(
|
|
self.apiclient,
|
|
ipaddressid=public_ip.ipaddress.id,
|
|
protocol='TCP',
|
|
cidrlist=[self.services["fwrule"]["cidr"]],
|
|
startport=self.services["fwrule"]["startport"],
|
|
endport=self.services["fwrule"]["endport"]
|
|
)
|
|
return public_ip
|
|
except Exception as e:
|
|
self.fail("Failed to acquire new public IP: %s" % e)
|
|
|
|
def get_router(self, account):
|
|
"""Returns a default router for account"""
|
|
|
|
routers = Router.list(self.apiclient,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid,
|
|
listall=True)
|
|
self.assertIsInstance(routers, list,
|
|
"List routers should return a valid response")
|
|
return routers[0]
|
|
|
|
def create_VPN(self, public_ip):
|
|
"""Creates VPN for the network"""
|
|
|
|
self.debug("Creating VPN with public IP: %s" % public_ip.ipaddress.id)
|
|
try:
|
|
# Assign VPN to Public IP
|
|
vpn = Vpn.create(self.apiclient,
|
|
self.public_ip.ipaddress.id,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid)
|
|
|
|
self.debug("Verifying the remote VPN access")
|
|
vpns = Vpn.list(self.apiclient,
|
|
publicipid=public_ip.ipaddress.id,
|
|
listall=True)
|
|
self.assertEqual(
|
|
isinstance(vpns, list),
|
|
True,
|
|
"List VPNs shall return a valid response"
|
|
)
|
|
return vpn
|
|
except Exception as e:
|
|
self.fail("Failed to create remote VPN access: %s" % e)
|
|
|
|
def try_ssh(self, ip_addr, hostnames):
|
|
try:
|
|
self.debug(
|
|
"SSH into (Public IP: %s)" % ip_addr)
|
|
|
|
# If Round Robin Algorithm is chosen,
|
|
# each ssh command should alternate between VMs
|
|
|
|
ssh_1 = SshClient(
|
|
ip_addr,
|
|
22,
|
|
self.virtual_machine.username,
|
|
self.virtual_machine.password
|
|
)
|
|
hostnames.append(ssh_1.execute("hostname")[0])
|
|
self.debug(hostnames)
|
|
except Exception as e:
|
|
self.fail("%s: SSH failed for VM with IP Address: %s" %
|
|
(e, ip_addr))
|
|
return hostnames
|
|
|
|
@attr(tags=["advanced", "advancedns"], required_hardware="true")
|
|
@attr(speed="slow")
|
|
def test_01_create_sticky_policy_default_values(self):
|
|
"""Test Configure stickiness policies with default values"""
|
|
|
|
# Validate the following
|
|
# 1. Create a LB rule with round robin. listLoadBalancerRules should
|
|
# show newly created load balancer rule.
|
|
# 2. Configure the Source based, app cookie and lb cookie based policy
|
|
# listLBStickinessPolicies should show newly created stickiness
|
|
|
|
self.debug("Creating a load balancing rule on IP: %s" %
|
|
self.public_ip.ipaddress.ipaddress)
|
|
|
|
lb_rule = self.create_LB_Rule(
|
|
self.public_ip,
|
|
network=self.get_Network(
|
|
self.account),
|
|
vmarray=[
|
|
self.virtual_machine,
|
|
self.virtual_machine_2])
|
|
|
|
#TODO: Add code to check the AppCookie and LbCookie Stickiness policies
|
|
methods = ["SourceBased"]
|
|
for method in methods:
|
|
self.debug("Creating stickiness policy for the LB rule: %s" %
|
|
lb_rule.id)
|
|
policies = self.configure_Stickiness_Policy(lb_rule, method=method)
|
|
|
|
policy = policies.stickinesspolicy[0]
|
|
|
|
self.debug("Policy: %s" % str(policy))
|
|
self.debug("Validating the stickiness policy")
|
|
self.validate_Stickiness_Policy(
|
|
lb_rule,
|
|
method,
|
|
self.public_ip.ipaddress.ipaddress)
|
|
self.debug("Deleting the stickiness policy for lb rule: %s" %
|
|
lb_rule.name)
|
|
self.delete_Stickiness_policy(policy, lb_rule)
|
|
return
|
|
|
|
@attr(tags=["advanced", "advancedns"], required_hardware="true")
|
|
@attr(speed="slow")
|
|
def test_02_create_sticky_policy_custom_values(self):
|
|
"""Test Configure stickiness policies with custom values"""
|
|
|
|
# Validate the following
|
|
# 1. Create a LB rule with roundrobin, leastconn and source.
|
|
# listLoadBalancerRules should show newly created load balancer rule
|
|
# 2. Configure the Source based, app cookie and lb cookie based policy
|
|
# with custom parameters
|
|
# listLBStickinessPolicies should show newly created stickiness
|
|
|
|
lb_methods = ["roundrobin", "leastconn", "source"]
|
|
|
|
configs = {"SourceBased": {"tablesize": '100k'}}
|
|
|
|
#TODO: Add code to check the AppCookie and LbCookie Stickiness policies
|
|
for lb_method in lb_methods:
|
|
self.debug("Creating a load balancing rule on IP %s and algo %s" %
|
|
(self.public_ip.ipaddress.ipaddress, lb_method))
|
|
|
|
services = self.services["lbrule"]
|
|
services["alg"] = lb_method
|
|
|
|
lb_rule = self.create_LB_Rule(
|
|
self.public_ip,
|
|
network=self.get_Network(
|
|
self.account),
|
|
vmarray=[
|
|
self.virtual_machine,
|
|
self.virtual_machine_2],
|
|
services=services)
|
|
|
|
for method, params in list(configs.items()):
|
|
self.debug("Creating stickiness policy for the LB rule: %s" %
|
|
lb_rule.id)
|
|
policies = self.configure_Stickiness_Policy(lb_rule,
|
|
method=method,
|
|
paramDict=params)
|
|
|
|
policy = policies.stickinesspolicy[0]
|
|
self.debug("Policy: %s" % str(policy))
|
|
|
|
self.debug("Validating the stickiness policy")
|
|
self.validate_Stickiness_Policy(
|
|
lb_rule,
|
|
method,
|
|
self.public_ip.ipaddress.ipaddress)
|
|
self.debug("Deleting the stickiness policy for lb rule: %s" %
|
|
lb_rule.name)
|
|
self.delete_Stickiness_policy(policy, lb_rule)
|
|
self.debug("Deleting the LB rule: %s" % lb_rule.name)
|
|
lb_rule.delete(self.apiclient)
|
|
return
|
|
|
|
@attr(tags=["advanced", "advancedns"])
|
|
@attr(speed="slow")
|
|
def test_03_supported_policies_by_network(self):
|
|
"""Test listnetworks response to check supported stickiness policies"""
|
|
|
|
# Validate the following
|
|
# 1. List networks for the account in advance network mode
|
|
# 2. List of supported sticky methods should be present under
|
|
# SupportedStickinessMethods tag
|
|
|
|
self.debug("List networks for account: %s" % self.account.name)
|
|
networks = Network.list(self.apiclient,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid,
|
|
listall=True)
|
|
|
|
self.assertIsInstance(networks,
|
|
list,
|
|
"List network should return a valid response")
|
|
network = networks[0]
|
|
self.debug("Network: %s" % network)
|
|
self.assertEqual(
|
|
hasattr(
|
|
network,
|
|
"SupportedStickinessMethods"),
|
|
True,
|
|
"Network should have SupportedStickinessMethods param")
|
|
|
|
self.assertEqual(hasattr(network, "LbCookie"),
|
|
True,
|
|
"Network should have LbCookie LB method param")
|
|
|
|
self.assertEqual(hasattr(network, "AppCookie"),
|
|
True,
|
|
"Network should have AppCookie LB method param")
|
|
|
|
self.assertEqual(hasattr(network, "SourceBased"),
|
|
True,
|
|
"Network should have SourceBased LB method param")
|
|
|
|
return
|
|
|
|
@attr(tags=["advanced", "advancedns"], required_hardware="true")
|
|
@attr(speed="slow")
|
|
def test_04_delete_lb_rule(self):
|
|
"""Test LB rule before/after stickiness policy creation"""
|
|
|
|
# Validate the following
|
|
# 1. Create a LB rule with roundrobin, leastconn and source.
|
|
# listLoadBalancerRules should show newly created load balancer rule
|
|
# 2. Delete the loadbalancer rule. Delete loadbalancer rule should be
|
|
# successful
|
|
# 3. Configure the Source based, app cookie and lb cookie based policy
|
|
# with custom parameters listLBStickinessPolicies should show newly
|
|
# created stickiness
|
|
# 4. Delete load balancer rule. Delete should be successful
|
|
|
|
lb_methods = ["roundrobin", "leastconn", "source"]
|
|
|
|
#TODO: Add code to check the AppCookie and LbCookie Stickiness policies
|
|
configs = {"SourceBased": {"tablesize": '100k'}}
|
|
for lb_method in lb_methods:
|
|
for method, params in list(configs.items()):
|
|
self.debug("Creating load balancing rule on IP %s & algo %s" %
|
|
(self.public_ip.ipaddress.ipaddress, lb_method))
|
|
|
|
services = self.services["lbrule"]
|
|
services["alg"] = lb_method
|
|
|
|
lb_rule = self.create_LB_Rule(
|
|
self.public_ip,
|
|
network=self.get_Network(
|
|
self.account),
|
|
vmarray=[
|
|
self.virtual_machine,
|
|
self.virtual_machine_2],
|
|
services=services)
|
|
self.debug(
|
|
"Deleting the LB rule before stickiness policy creation")
|
|
lb_rule.delete(self.apiclient)
|
|
|
|
with self.assertRaises(Exception):
|
|
LoadBalancerRule.list(self.apiclient,
|
|
id=lb_rule.id,
|
|
listall=True)
|
|
|
|
lb_rule = self.create_LB_Rule(
|
|
self.public_ip,
|
|
network=self.get_Network(
|
|
self.account),
|
|
vmarray=[
|
|
self.virtual_machine,
|
|
self.virtual_machine_2],
|
|
services=services)
|
|
self.debug("Creating stickiness policy for the LB rule: %s" %
|
|
lb_rule.id)
|
|
policies = self.configure_Stickiness_Policy(lb_rule,
|
|
method=method,
|
|
paramDict=params)
|
|
|
|
policy = policies.stickinesspolicy[0]
|
|
|
|
self.debug("Policy: %s" % str(policy))
|
|
self.debug("Validating the stickiness policy")
|
|
self.validate_Stickiness_Policy(
|
|
lb_rule,
|
|
method,
|
|
self.public_ip.ipaddress.ipaddress)
|
|
|
|
self.debug("Deleting the LB rule: %s" % lb_rule.name)
|
|
lb_rule.delete(self.apiclient)
|
|
with self.assertRaises(Exception):
|
|
LoadBalancerRule.list(self.apiclient, id=lb_rule.id)
|
|
return
|
|
|
|
@attr(tags=["advanced", "advancedns"], required_hardware="true")
|
|
@attr(speed="slow")
|
|
def test_05_error_alerts_after_create(self):
|
|
"""Test error/alerts after creating stickiness policy"""
|
|
|
|
# Validate the following
|
|
# 1. Create a LB rule with round Robin/Least connections/Source
|
|
# listLoadBalancerRules should show newly created load balancer rule
|
|
# 2. Configure the Stickiness policy to above created LB rule.
|
|
# listLBStickinessPolicies Api should show newly created stickiness
|
|
# 3. update & delete stickiness policy see error related to stickiness
|
|
# 4. No errors should be shown in the logs and alerts
|
|
|
|
lb_methods = ["roundrobin", "leastconn", "source"]
|
|
#TODO: Add code to check the AppCookie and LbCookie Stickiness policies
|
|
configs = {"SourceBased": {"tablesize": '100k'}}
|
|
for lb_method in lb_methods:
|
|
for method, params in list(configs.items()):
|
|
self.debug("Creating load balancing rule on IP %s & algo %s" %
|
|
(self.public_ip.ipaddress.ipaddress, lb_method))
|
|
|
|
services = self.services["lbrule"]
|
|
services["alg"] = lb_method
|
|
|
|
lb_rule = self.create_LB_Rule(
|
|
self.public_ip,
|
|
network=self.get_Network(
|
|
self.account),
|
|
vmarray=[
|
|
self.virtual_machine,
|
|
self.virtual_machine_2],
|
|
services=services)
|
|
|
|
self.debug("Creating stickiness policy for the LB rule: %s" %
|
|
lb_rule.id)
|
|
policies = self.configure_Stickiness_Policy(lb_rule,
|
|
method=method,
|
|
paramDict=params)
|
|
|
|
policy = policies.stickinesspolicy[0]
|
|
|
|
self.debug("Policy: %s" % str(policy))
|
|
self.debug("Validating the stickiness policy")
|
|
self.validate_Stickiness_Policy(
|
|
lb_rule,
|
|
method,
|
|
self.public_ip.ipaddress.ipaddress)
|
|
|
|
self.debug("Deleting the LB rule: %s" % lb_rule.name)
|
|
lb_rule.delete(self.apiclient)
|
|
|
|
with self.assertRaises(Exception):
|
|
LoadBalancerRule.list(self.apiclient,
|
|
id=lb_rule.id,
|
|
listall=True)
|
|
alerts = Alert.list(self.apiclient, keyword="stickiness",
|
|
listall=True)
|
|
self.debug(
|
|
"Create/update/delete should not produce any alert/error")
|
|
self.assertEqual(
|
|
alerts,
|
|
None,
|
|
"Create/update/delete should not produce any alert/error")
|
|
return
|
|
|
|
@attr(tags=["advanced", "advancedns"], required_hardware="true")
|
|
@attr(speed="slow")
|
|
def test_06_release_ip(self):
|
|
"""Test release public IP with stickiness policy"""
|
|
|
|
# 1. Configure load balancing rule. Listloadbalancerrule should list
|
|
# valid list
|
|
# 2. Create stickiness policy. liststickinesspolicy should return valid
|
|
# response
|
|
# 3. Release public Ip. liststickiness policy should return a valid
|
|
# response
|
|
|
|
lb_methods = ["roundrobin", "leastconn", "source"]
|
|
|
|
#TODO: Add code to check the AppCookie and LbCookie Stickiness policies
|
|
configs = {"SourceBased": {"tablesize": '100k'}}
|
|
|
|
for lb_method in lb_methods:
|
|
for method, params in list(configs.items()):
|
|
self.debug("Setting up environment - acquire public IP")
|
|
public_ip = self.acquire_Public_Ip()
|
|
|
|
self.debug(
|
|
"Creating a load balancing rule on IP %s and algo %s" %
|
|
(public_ip.ipaddress.ipaddress, lb_method))
|
|
|
|
services = self.services["lbrule"]
|
|
services["alg"] = lb_method
|
|
|
|
lb_rule = self.create_LB_Rule(
|
|
public_ip,
|
|
network=self.get_Network(
|
|
self.account),
|
|
vmarray=[
|
|
self.virtual_machine,
|
|
self.virtual_machine_2],
|
|
services=services)
|
|
|
|
policies = self.configure_Stickiness_Policy(lb_rule,
|
|
method=method,
|
|
paramDict=params)
|
|
policy = policies.stickinesspolicy[0]
|
|
|
|
self.debug("Policy: %s" % str(policy))
|
|
self.debug("Validating the stickiness policy")
|
|
self.validate_Stickiness_Policy(
|
|
lb_rule,
|
|
method,
|
|
public_ip.ipaddress.ipaddress)
|
|
|
|
self.debug("Releasing public Ip: %s" %
|
|
public_ip.ipaddress.ipaddress)
|
|
public_ip.delete(self.apiclient)
|
|
|
|
self.debug("Checking the response of liststickiness policies")
|
|
|
|
with self.assertRaises(Exception):
|
|
lb_rule.listStickyPolicies(self.apiclient,
|
|
lbruleid=lb_rule.id,
|
|
listall=True)
|
|
return
|
|
|
|
@attr(tags=["advanced", "advancedns"], required_hardware="true")
|
|
def test_07_delete_account(self):
|
|
"""Test Delete account and check the router and its rules"""
|
|
|
|
# Validate the following
|
|
# 1. create an account
|
|
# 2. using that account,create an instances
|
|
# 3. select the Source NAT IP and configure the stikiness policy
|
|
# 4. Delete account
|
|
# 5. The corresponding stikiness policy should be removed
|
|
# listLBStickinessPolicies Api shouldnot show deleted stikiness policy
|
|
|
|
self.debug("Creating LB rule for account: %s" %
|
|
self.account.name)
|
|
lb_rule = self.create_LB_Rule(
|
|
self.public_ip,
|
|
network=self.get_Network(
|
|
self.account),
|
|
vmarray=[
|
|
self.virtual_machine,
|
|
self.virtual_machine_2])
|
|
|
|
policies = self.configure_Stickiness_Policy(lb_rule, method="SourceBased")
|
|
policy = policies.stickinesspolicy[0]
|
|
|
|
self.debug("Policy: %s" % str(policy))
|
|
self.debug("Validating the stickiness policy")
|
|
self.validate_Stickiness_Policy(
|
|
lb_rule,
|
|
"SourceBased",
|
|
self.public_ip.ipaddress.ipaddress)
|
|
|
|
# removing account from cleanup list as we're deleting account
|
|
self.cleanup.pop()
|
|
self.debug("Deleting account: %s" % self.account.name)
|
|
|
|
try:
|
|
self.account.delete(self.apiclient)
|
|
except Exception as e:
|
|
self.fail("Failed to delete account: %s" % e)
|
|
self.debug("Checking the response of liststickiness policies")
|
|
|
|
with self.assertRaises(Exception):
|
|
lb_rule.listStickyPolicies(self.apiclient,
|
|
lbruleid=lb_rule.id,
|
|
listall=True)
|
|
return
|
|
|
|
@attr(tags=["advanced", "advancedns"], required_hardware="true")
|
|
def test_08_create_policy_router_stopped(self):
|
|
"""Test verify create stickiness policy when router is stopped state"""
|
|
|
|
# Validate the following
|
|
# 1. stop the router
|
|
# 2. create stikiness policy from UI
|
|
# 3. start the router. listLBStickinessPolicies Api should show created
|
|
# stikiness policy
|
|
|
|
self.debug("Creating LB rule for account: %s" % self.account.name)
|
|
lb_rule = self.create_LB_Rule(
|
|
self.public_ip,
|
|
network=self.get_Network(
|
|
self.account),
|
|
vmarray=[
|
|
self.virtual_machine,
|
|
self.virtual_machine_2])
|
|
|
|
self.debug("Fetching routers for the account: %s" %
|
|
self.account.name)
|
|
router = self.get_router(self.account)
|
|
|
|
self.debug("Stopping the router: %s" % router.name)
|
|
Router.stop(self.apiclient, id=router.id)
|
|
|
|
policies = self.configure_Stickiness_Policy(lb_rule, method="SourceBased")
|
|
policy = policies.stickinesspolicy[0]
|
|
|
|
self.debug("Starting the router: %s" % router.name)
|
|
Router.start(self.apiclient, id=router.id)
|
|
|
|
response = verifyRouterState(self.apiclient,
|
|
router.id,
|
|
"running")
|
|
self.assertEqual(response[0], PASS, response[1])
|
|
|
|
self.debug("Policy: %s" % str(policy))
|
|
self.debug("Validating the stickiness policy")
|
|
self.validate_Stickiness_Policy(
|
|
lb_rule,
|
|
"SourceBased",
|
|
self.public_ip.ipaddress.ipaddress)
|
|
return
|
|
|
|
@attr(tags=["advanced", "advancedns"], required_hardware="true")
|
|
def test_09_create_policy_router_destroy(self):
|
|
"""Test check the stickiness policy rules after destroying router"""
|
|
|
|
# Validate the following
|
|
# 1. create an account
|
|
# 2. using that account,create an instances
|
|
# 3. select the Source NAT IP and configure the stikiness policy
|
|
# 4. destroy the router.
|
|
|
|
self.debug("Creating LB rule for account: %s" % self.account.name)
|
|
lb_rule = self.create_LB_Rule(
|
|
self.public_ip,
|
|
network=self.get_Network(
|
|
self.account),
|
|
vmarray=[
|
|
self.virtual_machine,
|
|
self.virtual_machine_2])
|
|
|
|
self.debug("Fetching routers for the account: %s" %
|
|
self.account.name)
|
|
router = self.get_router(self.account)
|
|
|
|
policies = self.configure_Stickiness_Policy(lb_rule, method="SourceBased")
|
|
policy = policies.stickinesspolicy[0]
|
|
|
|
self.debug("Policy: %s" % str(policy))
|
|
self.debug("Validating the stickiness policy")
|
|
self.validate_Stickiness_Policy(
|
|
lb_rule,
|
|
"SourceBased",
|
|
self.public_ip.ipaddress.ipaddress)
|
|
|
|
self.debug("Destroying the router: %s" % router.name)
|
|
Router.destroy(self.apiclient, id=router.id)
|
|
|
|
return
|
|
|
|
@attr(tags=["advanced", "advancedns"], required_hardware="true")
|
|
def test_10_create_policy_enable_disable_vpn(self):
|
|
"""Test enable/disable the VPN after applying sticky policy rules"""
|
|
|
|
# Validate the following
|
|
# 1. create an account
|
|
# 2. using that account,create an instances
|
|
# 3. select the Source NAT IP and configure the stikiness policy
|
|
# 4. enable /disable the VPN. It should not impact the created rules
|
|
# listLBStickinessPolicies Api should show created stikiness policy
|
|
|
|
self.debug("Creating LB rule for account: %s" % self.account.name)
|
|
lb_rule = self.create_LB_Rule(
|
|
self.public_ip,
|
|
network=self.get_Network(
|
|
self.account),
|
|
vmarray=[
|
|
self.virtual_machine,
|
|
self.virtual_machine_2])
|
|
|
|
policies = self.configure_Stickiness_Policy(lb_rule, method="SourceBased")
|
|
policy = policies.stickinesspolicy[0]
|
|
|
|
self.debug("Policy: %s" % str(policy))
|
|
self.debug("Validating the stickiness policy")
|
|
self.validate_Stickiness_Policy(
|
|
lb_rule,
|
|
"SourceBased",
|
|
self.public_ip.ipaddress.ipaddress)
|
|
|
|
self.debug("Enabling VPN on Public Ip: %s" %
|
|
self.public_ip.ipaddress.ipaddress)
|
|
self.create_VPN(self.public_ip)
|
|
|
|
self.debug("Validating the stickiness policy after enabling VPN")
|
|
self.validate_Stickiness_Policy(
|
|
lb_rule,
|
|
"SourceBased",
|
|
self.public_ip.ipaddress.ipaddress)
|
|
return
|
|
|
|
@attr(tags=["advanced", "advancedns"], required_hardware="false")
|
|
def test_11_invalid_params(self):
|
|
"""Test verify functionality syncronous and asyncronous validations"""
|
|
|
|
# Validate the following
|
|
# verify the validation while creating or attaching stikiness policy
|
|
# by doing the following scenaios
|
|
# * by passing the Invlaid parameter
|
|
# * Invalid method name
|
|
# * required parameter not present
|
|
# * passing invalid values to valid parameters.
|
|
|
|
self.debug("Creating LB rule for account: %s" % self.account.name)
|
|
lb_rule = self.create_LB_Rule(
|
|
self.public_ip,
|
|
network=self.get_Network(
|
|
self.account),
|
|
vmarray=[
|
|
self.virtual_machine,
|
|
self.virtual_machine_2])
|
|
|
|
self.debug("Creating stickiness policy with invalid method")
|
|
with self.assertRaises(Exception):
|
|
self.configure_Stickiness_Policy(lb_rule, method="InvalidMethod")
|
|
|
|
self.debug("Creating stickiness policy with invalid params")
|
|
with self.assertRaises(Exception):
|
|
self.configure_Stickiness_Policy(lb_rule, method="LbCookie",
|
|
params={"Test": 10})
|
|
|
|
self.debug("Passing invalid parameter")
|
|
with self.assertRaises(Exception):
|
|
cmd = createLBStickinessPolicy.createLBStickinessPolicyCmd()
|
|
cmd.lbruleid = lb_rule.id
|
|
cmd.method = "LbCookie"
|
|
cmd.name = "LbCookie"
|
|
self.apiclient.createLBStickinessPolicy(cmd)
|
|
|
|
self.debug("Creating stickiness policy not passing required param")
|
|
with self.assertRaises(Exception):
|
|
cmd = createLBStickinessPolicy.createLBStickinessPolicyCmd()
|
|
cmd.lbruleid = lb_rule.id
|
|
cmd.name = "LbCookie"
|
|
self.apiclient.createLBStickinessPolicy(cmd)
|
|
|
|
return
|