etan39
/
cloudstack
mirror of https://github.com/apache/cloudstack.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Automation Tests for HA Proxy Stickiness 

Reviewed-by: Suresh Sadhu <Suresh.Sadhu@citrix.com>
Signed-off-by: Prasanna Santhanam <tsp@apache.org>
This commit is contained in:
Girish Shilamkar 2013-10-10 02:21:47 -04:00 committed by Prasanna Santhanam
parent 76dbb64cf9
commit 56ccb2835b
1 changed files with 874 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

874
test/integration/component/test_haproxy.py Normal file
View File

@ -0,0 +1,874 @@
# 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.integration.lib.base import (
Account,
ServiceOffering,
VirtualMachine,
PublicIPAddress,
Network,
LoadBalancerRule,
Alert,
Router,
Vpn,
NATRule
)
from marvin.integration.lib.common import (get_domain,
get_zone,
get_template,
cleanup_resources,
random_gen
)
from marvin.cloudstackAPI import createLBStickinessPolicy
from marvin.remoteSSHClient import remoteSSHClient
class Services:
"""Test VPN users Services
"""
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",
},
"service_offering": {
"name": "Tiny Instance",
"displaytext": "Tiny Instance",
"cpunumber": 1,
"cpuspeed": 100, # in MHz
"memory": 128, # In MBs
},
"disk_offering": {
"displaytext": "Small Disk Offering",
"name": "Small Disk Offering",
"disksize": 1
},
"virtual_machine": {
"displayname": "TestVM",
"username": "root",
"password": "password",
"ssh_port": 22,
"hypervisor": 'XenServer',
"privateport": 22,
"publicport": 22,
"protocol": 'TCP',
},
"vpn_user": {
"username": "test",
"password": "test",
},
"natrule": {
"privateport": 22,
"publicport": 22,
"protocol": "TCP",
"username":"root",
"password": "password"
},
"network": {
"name": "Test Network",
"displaytext": "Test Network",
"netmask": '255.255.255.0'
},
"lbrule": {
"name": "SSH",
"alg": "roundrobin",
# Algorithm used for load balancing
"privateport": 22,
"publicport": 2222,
"openfirewall": False,
"startport": 22,
"endport": 2222,
"protocol": "TCP",
"cidrlist": '0.0.0.0/0',
},
"ostype": 'CentOS 5.3 (64-bit)',
"sleep": 60,
"timeout": 10,
"mode": 'advanced',
# Networking mode: Advanced, Basic
}
class TestHAProxyStickyness(cloudstackTestCase):
@classmethod
def setUpClass(cls):
cls.api_client = super(TestHAProxyStickyness,
cls).getClsTestClient().getApiClient()
cls.services = Services().services
# 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.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)
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"]
)
NATRule.create(
self.apiclient,
self.virtual_machine,
self.services["natrule"],
ipaddressid=self.public_ip.ipaddress.id
)
self.cleanup = [self.account, ]
return
def tearDown(self):
try:
# Clean up, terminate the created instance, volumes and snapshots
cleanup_resources(self.apiclient, self.cleanup)
pass
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"]
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)
self.debug("Configuring NAT rule for the acquired public ip")
NATRule.create(
self.apiclient,
self.virtual_machine,
self.services["natrule"],
ipaddressid=public_ip.ipaddress.id
)
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 repsonse")
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 NAT Rule (Public IP: %s)" % ip_addr)
# If Round Robin Algorithm is chosen,
# each ssh command should alternate between VMs
ssh_1 = remoteSSHClient(
ip_addr,
22,
self.services["natrule"]["username"],
self.services["natrule"]["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"])
@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])
methods = ["SourceBased", "AppCookie", "LBCookie"]
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"])
@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'},
"AppCookie": {"request-learn": "true"},
"LBCookie": {"nocache": "true"}}
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 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"])
@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"]
configs = {"SourceBased": {"tablesize": '100k'},
"AppCookie": {"request-learn": "true"},
"LBCookie": {"nocache": "true"}}
for lb_method in lb_methods:
for method, params in 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"])
@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"]
configs = {"SourceBased": {"tablesize": '100k'},
"AppCookie": {"request-learn": "true"},
"LBCookie": {"nocache": "true"}}
for lb_method in lb_methods:
for method, params in 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"])
@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"]
configs = {"SourceBased": {"tablesize": '100k'},
"AppCookie": {"request-learn": "true"},
"LBCookie": {"nocache": "true"}}
for lb_method in lb_methods:
for method, params in 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"])
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="LbCookie")
policy = policies.stickinesspolicy[0]
self.debug("Policy: %s" % str(policy))
self.debug("Validating the stickiness policy")
self.validate_Stickiness_Policy(lb_rule, "LbCookie", 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"])
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="LbCookie")
policy = policies.stickinesspolicy[0]
self.debug("Starting the router: %s" % router.name)
Router.start(self.apiclient, id=router.id)
self.debug("Policy: %s" % str(policy))
self.debug("Validating the stickiness policy")
self.validate_Stickiness_Policy(lb_rule, "LbCookie", self.public_ip.ipaddress.ipaddress)
return
@attr(tags=["advanced", "advancedns"])
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="LbCookie")
policy = policies.stickinesspolicy[0]
self.debug("Policy: %s" % str(policy))
self.debug("Validating the stickiness policy")
self.validate_Stickiness_Policy(lb_rule, "LbCookie", 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"])
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 ceated 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="LbCookie")
policy = policies.stickinesspolicy[0]
self.debug("Policy: %s" % str(policy))
self.debug("Validating the stickiness policy")
self.validate_Stickiness_Policy(lb_rule, "LbCookie", 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, "LbCookie", self.public_ip.ipaddress.ipaddress)
return
@attr(tags=["advanced", "advancedns"])
def test_11_invalid_params(self):
"""Test verfify 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 paramters.
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