You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by ta...@apache.org on 2014/11/14 12:10:09 UTC
git commit: updated refs/heads/4.5 to 2bfddc5
Repository: cloudstack
Updated Branches:
refs/heads/4.5 a782495c6 -> 2bfddc55b
CLOUDSTACK-7860: test_eip_elb.py - Move Netscler info out of the test case. Read it from config. Fix attribute error. Fix pep8 issues. Fix imports.
Signed-off-by: SrikanteswaraRao Talluri <ta...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/cloudstack/repo
Commit: http://git-wip-us.apache.org/repos/asf/cloudstack/commit/2bfddc55
Tree: http://git-wip-us.apache.org/repos/asf/cloudstack/tree/2bfddc55
Diff: http://git-wip-us.apache.org/repos/asf/cloudstack/diff/2bfddc55
Branch: refs/heads/4.5
Commit: 2bfddc55b8e886323173fe56d6735472a9da5f69
Parents: a782495
Author: Gaurav Aradhye <ga...@clogeny.com>
Authored: Thu Nov 13 17:54:46 2014 +0530
Committer: SrikanteswaraRao Talluri <ta...@apache.org>
Committed: Fri Nov 14 16:39:55 2014 +0530
----------------------------------------------------------------------
test/integration/component/test_eip_elb.py | 1518 ++++++++++++-----------
1 file changed, 771 insertions(+), 747 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/2bfddc55/test/integration/component/test_eip_elb.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_eip_elb.py b/test/integration/component/test_eip_elb.py
index c35ea9b..76987b9 100644
--- a/test/integration/component/test_eip_elb.py
+++ b/test/integration/component/test_eip_elb.py
@@ -5,9 +5,9 @@
# 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
@@ -17,75 +17,83 @@
""" P1 tests for elastic load balancing and elastic IP
"""
-#Import Local Modules
-import marvin
+# Import Local Modules
from nose.plugins.attrib import attr
-from marvin.cloudstackTestCase import *
-from marvin.cloudstackAPI import *
-from marvin.lib.utils import *
-from marvin.lib.base import *
-from marvin.lib.common import *
+from marvin.cloudstackTestCase import cloudstackTestCase, unittest
+from marvin.cloudstackAPI import (authorizeSecurityGroupIngress,
+ disassociateIpAddress,
+ deleteLoadBalancerRule)
+from marvin.lib.utils import cleanup_resources
+from marvin.lib.base import (Account,
+ PublicIPAddress,
+ VirtualMachine,
+ Network,
+ LoadBalancerRule,
+ SecurityGroup,
+ ServiceOffering,
+ StaticNATRule,
+ PublicIpRange)
+from marvin.lib.common import (get_zone,
+ get_domain,
+ get_template)
from marvin.sshClient import SshClient
-import datetime
+import time
class Services:
+
"""Test elastic load balancing and elastic IP
"""
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
- },
- "lbrule": {
- "name": "SSH",
- "alg": "roundrobin",
- # Algorithm used for load balancing
- "privateport": 22,
- "publicport": 22,
- "openfirewall": False,
- },
- "natrule": {
- "privateport": 22,
- "publicport": 22,
- "protocol": "TCP"
- },
- "virtual_machine": {
- "displayname": "Test VM",
- "username": "root",
- "password": "password",
- "ssh_port": 22,
- "hypervisor": 'XenServer',
- # Hypervisor type should be same as
- # hypervisor type of cluster
- "privateport": 22,
- "publicport": 22,
- "protocol": 'TCP',
- },
- "netscaler": {
- "ipaddress": '10.147.40.100',
- "username": 'nsroot',
- "password": 'nsroot'
- },
- "ostype": 'CentOS 5.3 (64-bit)',
- # Cent OS 5.3 (64 bit)
- "sleep": 60,
- "timeout": 10,
- }
+ "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
+ },
+ "lbrule": {
+ "name": "SSH",
+ "alg": "roundrobin",
+ # Algorithm used for load balancing
+ "privateport": 22,
+ "publicport": 22,
+ "openfirewall": False,
+ },
+ "natrule": {
+ "privateport": 22,
+ "publicport": 22,
+ "protocol": "TCP"
+ },
+ "virtual_machine": {
+ "displayname": "Test VM",
+ "username": "root",
+ "password": "password",
+ "ssh_port": 22,
+ "hypervisor": 'XenServer',
+ # Hypervisor type should be same as
+ # hypervisor type of cluster
+ "privateport": 22,
+ "publicport": 22,
+ "protocol": 'TCP',
+ },
+ "ostype": 'CentOS 5.3 (64-bit)',
+ # Cent OS 5.3 (64 bit)
+ "sleep": 60,
+ "timeout": 10,
+ }
+
class TestEIP(cloudstackTestCase):
@@ -95,74 +103,85 @@ class TestEIP(cloudstackTestCase):
cls.api_client = cls.testClient.getApiClient()
cls.services = Services().services
+
+ try:
+ cls.services["netscaler"] = cls.config.__dict__[
+ "netscalerDevice"].__dict__
+ except KeyError:
+ raise unittest.SkipTest("Please make sure you have included netscalerDevice\
+ dict in your config file (keys - ipaddress, username,\
+ password")
+ except Exception as e:
+ raise unittest.SkipTest(e)
+
# Get Zone, Domain and templates
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
cls.template = get_template(
- cls.api_client,
- cls.zone.id,
- cls.services["ostype"]
- )
+ 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.api_client,
+ cls.services["service_offering"]
+ )
cls.account = Account.create(
- cls.api_client,
- cls.services["account"],
- admin=True,
- domainid=cls.domain.id
- )
+ cls.api_client,
+ cls.services["account"],
+ admin=True,
+ domainid=cls.domain.id
+ )
# Spawn an instance
cls.virtual_machine = VirtualMachine.create(
- cls.api_client,
- cls.services["virtual_machine"],
- accountid=cls.account.name,
- domainid=cls.account.domainid,
- serviceofferingid=cls.service_offering.id
- )
+ cls.api_client,
+ cls.services["virtual_machine"],
+ accountid=cls.account.name,
+ domainid=cls.account.domainid,
+ serviceofferingid=cls.service_offering.id
+ )
networks = Network.list(
- cls.api_client,
- zoneid=cls.zone.id,
- listall=True
- )
+ cls.api_client,
+ zoneid=cls.zone.id,
+ listall=True
+ )
if isinstance(networks, list):
# Basic zone has only one network i.e Basic network
cls.guest_network = networks[0]
else:
raise Exception(
"List networks returned empty response for zone: %s" %
- cls.zone.id)
+ cls.zone.id)
ip_addrs = PublicIPAddress.list(
- cls.api_client,
- associatednetworkid=cls.guest_network.id,
- isstaticnat=True,
- account=cls.account.name,
- domainid=cls.account.domainid,
- listall=True
- )
+ cls.api_client,
+ associatednetworkid=cls.guest_network.id,
+ isstaticnat=True,
+ account=cls.account.name,
+ domainid=cls.account.domainid,
+ listall=True
+ )
if isinstance(ip_addrs, list):
cls.source_nat = ip_addrs[0]
- print "source_nat ipaddress : ", cls.source_nat
+ print "source_nat ipaddress : ", cls.source_nat.ipaddress
else:
raise Exception(
- "No Source NAT IP found for guest network: %s" %
- cls.guest_network.id)
+ "No Source NAT IP found for guest network: %s" %
+ cls.guest_network.id)
cls._cleanup = [
- cls.account,
- cls.service_offering,
- ]
+ cls.account,
+ cls.service_offering,
+ ]
return
@classmethod
def tearDownClass(cls):
try:
- #Cleanup resources used
+ # Cleanup resources used
cleanup_resources(cls.api_client, cls._cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
@@ -176,19 +195,17 @@ class TestEIP(cloudstackTestCase):
def tearDown(self):
try:
- #Clean up, terminate the created network offerings
+ # Clean up, terminate the created network offerings
cleanup_resources(self.apiclient, self.cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
-
- @attr(tags = ["eip"])
+ @attr(tags=["eip"])
def test_01_eip_by_deploying_instance(self):
"""Test EIP by deploying an instance
"""
-
# Validate the following
# 1. Instance gets an IP from GUEST IP range.
# 2. One IP from EIP pool is taken and configured on NS
@@ -200,62 +217,57 @@ class TestEIP(cloudstackTestCase):
self.debug("Fetching public network IP range for public network")
ip_ranges = PublicIpRange.list(
- self.apiclient,
- zoneid=self.zone.id,
- forvirtualnetwork=True
- )
+ self.apiclient,
+ zoneid=self.zone.id,
+ forvirtualnetwork=True
+ )
self.assertEqual(
- isinstance(ip_ranges, list),
- True,
- "Public IP range should return a valid range"
- )
+ isinstance(ip_ranges, list),
+ True,
+ "Public IP range should return a valid range"
+ )
# Guest network can have multiple IP ranges. In that case, split IP
# address and then compare the values
for ip_range in ip_ranges:
self.debug("IP range: %s - %s" % (
- ip_range.startip,
- ip_range.endip
- ))
+ ip_range.startip,
+ ip_range.endip
+ ))
start_ip_list = ip_range.startip.split(".")
end_ip_list = ip_range.endip.split(".")
source_nat_list = self.source_nat.ipaddress.split(".")
- flag = True
- for i in range(3):
- if int(start_ip_list[i]) != int(source_nat_list[i]):
- flag = False
-
self.assertGreaterEqual(
int(source_nat_list[3]),
int(start_ip_list[3]),
"The NAT should be greater/equal to start IP of guest network"
- )
+ )
self.assertLessEqual(
int(source_nat_list[3]),
int(end_ip_list[3]),
"The NAT should be less/equal to start IP of guest network"
- )
+ )
# Verify listSecurity groups response
security_groups = SecurityGroup.list(
- self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid
- )
+ self.apiclient,
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
self.assertEqual(
- isinstance(security_groups, list),
- True,
- "Check for list security groups response"
- )
+ isinstance(security_groups, list),
+ True,
+ "Check for list security groups response"
+ )
self.assertEqual(
- len(security_groups),
- 1,
- "Check List Security groups response"
- )
+ len(security_groups),
+ 1,
+ "Check List Security groups response"
+ )
self.debug("List Security groups response: %s" %
- str(security_groups))
+ str(security_groups))
security_group = security_groups[0]
@@ -271,7 +283,7 @@ class TestEIP(cloudstackTestCase):
cmd.endport = 22
cmd.cidrlist = '0.0.0.0/0'
self.apiclient.authorizeSecurityGroupIngress(cmd)
-#COMMENTED:
+# COMMENTED:
# try:
# self.debug("SSH into VM: %s" % self.virtual_machine.ssh_ip)
# ssh = self.virtual_machine.get_ssh_client(
@@ -282,80 +294,77 @@ class TestEIP(cloudstackTestCase):
# )
# Fetch details from user_ip_address table in database
self.debug(
- "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \
- % self.source_nat.ipaddress)
+ "select is_system, one_to_one_nat from user_ip_address\
+ where public_ip_address='%s';" %
+ self.source_nat.ipaddress)
qresultset = self.dbclient.execute(
- "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \
- % self.source_nat.ipaddress
- )
+ "select is_system, one_to_one_nat from user_ip_address\
+ where public_ip_address='%s';" %
+ self.source_nat.ipaddress)
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "Check DB Query result set"
- )
+ len(qresultset),
+ 0,
+ "Check DB Query result set"
+ )
qresult = qresultset[0]
self.assertEqual(
- qresult[0],
- 1,
- "user_ip_address.is_system value should be 1 for static NAT"
- )
+ qresult[0],
+ 1,
+ "user_ip_address.is_system value should be 1 for static NAT"
+ )
self.assertEqual(
- qresult[1],
- 1,
- "user_ip_address.one_to_one_nat value should be 1 for static NAT"
- )
+ qresult[1],
+ 1,
+ "user_ip_address.one_to_one_nat value should be 1 for static NAT"
+ )
self.debug("SSH into netscaler: %s" %
- self.services["netscaler"]["ipaddress"])
- try:
- ssh_client = SshClient(
- self.services["netscaler"]["ipaddress"],
- 22,
- self.services["netscaler"]["username"],
- self.services["netscaler"]["password"],
- )
- self.debug("command: show ip")
- res = ssh_client.execute("show ip")
- result = str(res)
- self.debug("Output: %s" % result)
+ self.services["netscaler"]["ipaddress"])
+
+ ssh_client = SshClient(
+ self.services["netscaler"]["ipaddress"],
+ 22,
+ self.services["netscaler"]["username"],
+ self.services["netscaler"]["password"],
+ )
+ self.debug("command: show ip")
+ res = ssh_client.execute("show ip")
+ result = str(res)
+ self.debug("Output: %s" % result)
- self.assertEqual(
- result.count(self.source_nat.ipaddress),
- 1,
- "One IP from EIP pool should be taken and configured on NS"
- )
-
- self.debug("Command:show inat")
- res = ssh_client.execute("show inat")
+ self.assertEqual(
+ result.count(self.source_nat.ipaddress),
+ 1,
+ "One IP from EIP pool should be taken and configured on NS"
+ )
- result = str(res)
- self.debug("Output: %s" % result)
+ self.debug("Command:show inat")
+ res = ssh_client.execute("show inat")
- self.assertEqual(
- result.count("NAME: Cloud-Inat-%s" % self.source_nat.ipaddress),
- 1,
- "User source IP should be enabled for INAT service"
- )
+ result = str(res)
+ self.debug("Output: %s" % result)
- except Exception as e:
- self.fail("SSH Access failed for %s: %s" % \
- (self.services["netscaler"]["ipaddress"], e))
+ self.assertEqual(
+ result.count(
+ "NAME: Cloud-Inat-%s" %
+ self.source_nat.ipaddress),
+ 1,
+ "User source IP should be enabled for INAT service")
return
- @attr(tags = ["eip"])
+ @attr(tags=["eip"])
def test_02_acquire_ip_enable_static_nat(self):
"""Test associate new IP and enable static NAT for new IP and the VM
"""
-
# Validate the following
# 1. user_ip_address.is_system = 0 & user_ip_address.one_to_one_nat=1
# 2. releases default EIP whose user_ip_address.is_system=1
@@ -369,84 +378,86 @@ class TestEIP(cloudstackTestCase):
self.debug("Acquiring new IP for network: %s" % self.guest_network.id)
public_ip = PublicIPAddress.create(
- self.apiclient,
- accountid=self.account.name,
- zoneid=self.zone.id,
- domainid=self.account.domainid,
- services=self.services["virtual_machine"]
- )
+ self.apiclient,
+ accountid=self.account.name,
+ zoneid=self.zone.id,
+ domainid=self.account.domainid,
+ services=self.services["virtual_machine"]
+ )
self.debug("IP address: %s is acquired by network: %s" % (
- public_ip.ipaddress,
- self.guest_network.id))
+ public_ip.ipaddress.ipaddress,
+ self.guest_network.id))
self.debug("Enabling static NAT for IP Address: %s" %
- public_ip.ipaddress)
+ public_ip.ipaddress.ipaddress)
StaticNATRule.enable(
- self.apiclient,
- ipaddressid=public_ip.ipaddress.id,
- virtualmachineid=self.virtual_machine.id
- )
+ self.apiclient,
+ ipaddressid=public_ip.ipaddress.id,
+ virtualmachineid=self.virtual_machine.id
+ )
# Fetch details from user_ip_address table in database
self.debug(
- "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \
- % public_ip.ipaddress)
+ "select is_system, one_to_one_nat from user_ip_address\
+ where public_ip_address='%s';" %
+ public_ip.ipaddress.ipaddress)
qresultset = self.dbclient.execute(
- "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \
- % public_ip.ipaddress
- )
+ "select is_system, one_to_one_nat from user_ip_address\
+ where public_ip_address='%s';" %
+ public_ip.ipaddress.ipaddress)
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "Check DB Query result set"
- )
+ len(qresultset),
+ 0,
+ "Check DB Query result set"
+ )
qresult = qresultset[0]
self.assertEqual(
- qresult[0],
- 0,
- "user_ip_address.is_system value should be 0 for new IP"
- )
+ qresult[0],
+ 0,
+ "user_ip_address.is_system value should be 0 for new IP"
+ )
self.assertEqual(
- qresult[1],
- 1,
- "user_ip_address.one_to_one_nat value should be 1 for static NAT"
- )
+ qresult[1],
+ 1,
+ "user_ip_address.one_to_one_nat value should be 1 for static NAT"
+ )
self.debug(
- "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \
- % self.source_nat.ipaddress)
+ "select is_system, one_to_one_nat from user_ip_address\
+ where public_ip_address='%s';" %
+ self.source_nat.ipaddress)
qresultset = self.dbclient.execute(
- "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \
- % self.source_nat.ipaddress
- )
+ "select is_system, one_to_one_nat from user_ip_address\
+ where public_ip_address='%s';" %
+ self.source_nat.ipaddress)
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "Check DB Query result set"
- )
+ len(qresultset),
+ 0,
+ "Check DB Query result set"
+ )
qresult = qresultset[0]
self.assertEqual(
- qresult[0],
- 0,
- "user_ip_address.is_system value should be 0 old source NAT"
- )
+ qresult[0],
+ 0,
+ "user_ip_address.is_system value should be 0 old source NAT"
+ )
# try:
# self.debug("SSH into VM: %s" % public_ip.ipaddress)
@@ -458,24 +469,24 @@ class TestEIP(cloudstackTestCase):
# )
self.debug("SSH into netscaler: %s" %
- self.services["netscaler"]["ipaddress"])
+ self.services["netscaler"]["ipaddress"])
try:
ssh_client = SshClient(
- self.services["netscaler"]["ipaddress"],
- 22,
- self.services["netscaler"]["username"],
- self.services["netscaler"]["password"],
- )
+ self.services["netscaler"]["ipaddress"],
+ 22,
+ self.services["netscaler"]["username"],
+ self.services["netscaler"]["password"],
+ )
self.debug("command: show ip")
res = ssh_client.execute("show ip")
result = str(res)
self.debug("Output: %s" % result)
self.assertEqual(
- result.count(public_ip.ipaddress),
- 1,
- "One IP from EIP pool should be taken and configured on NS"
- )
+ result.count(public_ip.ipaddress.ipaddress),
+ 1,
+ "One IP from EIP pool should be taken and configured on NS"
+ )
self.debug("Command:show inat")
res = ssh_client.execute("show inat")
@@ -484,22 +495,22 @@ class TestEIP(cloudstackTestCase):
self.debug("Output: %s" % result)
self.assertEqual(
- result.count("NAME: Cloud-Inat-%s" % public_ip.ipaddress),
- 1,
- "User source IP should be enabled for INAT service"
- )
+ result.count(
+ "NAME: Cloud-Inat-%s" %
+ public_ip.ipaddress.ipaddress),
+ 1,
+ "User source IP should be enabled for INAT service")
except Exception as e:
- self.fail("SSH Access failed for %s: %s" % \
+ self.fail("SSH Access failed for %s: %s" %
(self.services["netscaler"]["ipaddress"], e))
return
- @attr(tags = ["eip"])
+ @attr(tags=["eip"])
def test_03_disable_static_nat(self):
"""Test disable static NAT and release EIP acquired
"""
-
# Validate the following
# 1. Disable static NAT. Disables one-to-one NAT and releases EIP
# whose user_ip_address.is_system=0
@@ -513,53 +524,56 @@ class TestEIP(cloudstackTestCase):
self.debug(
"Fetching static NAT for VM: %s" % self.virtual_machine.name)
ip_addrs = PublicIPAddress.list(
- self.api_client,
- associatednetworkid=self.guest_network.id,
- isstaticnat=True,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ associatednetworkid=self.guest_network.id,
+ isstaticnat=True,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(ip_addrs, list),
- True,
- "List Public IP address should return valid IP address for network"
- )
+ isinstance(ip_addrs, list),
+ True,
+ "List Public IP address should return valid IP address for network"
+ )
static_nat = ip_addrs[0]
self.debug("Static NAT for VM: %s is: %s" % (
- self.virtual_machine.name,
- static_nat.ipaddress
- ))
+ self.virtual_machine.name,
+ static_nat.ipaddress
+ ))
# Fetch details from user_ip_address table in database
self.debug(
- "select is_system from user_ip_address where public_ip_address='%s';" \
- % static_nat.ipaddress)
+ "select is_system from user_ip_address where\
+ public_ip_address='%s';" %
+ static_nat.ipaddress)
qresultset = self.dbclient.execute(
- "select is_system from user_ip_address where public_ip_address='%s';" \
- % static_nat.ipaddress
- )
+ "select is_system from user_ip_address where\
+ public_ip_address='%s';" %
+ static_nat.ipaddress)
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "Check DB Query result set"
- )
+ len(qresultset),
+ 0,
+ "Check DB Query result set"
+ )
qresult = qresultset[0]
self.assertEqual(
- qresult[0],
- 0,
- "user_ip_address.is_system value should be 0"
- )
+ qresult[0],
+ 0,
+ "user_ip_address.is_system value should be 0"
+ )
- self.debug("Disassociate Static NAT: %s" % static_nat.ipaddress)
+ self.debug(
+ "Disassociate Static NAT: %s" %
+ static_nat.ipaddress)
cmd = disassociateIpAddress.disassociateIpAddressCmd()
cmd.id = static_nat.id
self.apiclient.disassociateIpAddress(cmd)
@@ -569,85 +583,86 @@ class TestEIP(cloudstackTestCase):
# Fetch details from user_ip_address table in database
self.debug(
- "select state from user_ip_address where public_ip_address='%s';" \
- % static_nat.ipaddress)
+ "select state from user_ip_address where public_ip_address='%s';"
+ % static_nat.ipaddress)
qresultset = self.dbclient.execute(
- "select state from user_ip_address where public_ip_address='%s';" \
- % static_nat.ipaddress
- )
+ "select state from user_ip_address where public_ip_address='%s';"
+ % static_nat.ipaddress
+ )
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "Check DB Query result set"
- )
+ len(qresultset),
+ 0,
+ "Check DB Query result set"
+ )
qresult = qresultset[0]
self.assertEqual(
- qresult[0],
- "Free",
- "Ip should be marked as Free after disassociate IP"
- )
+ qresult[0],
+ "Free",
+ "Ip should be marked as Free after disassociate IP"
+ )
self.debug(
"Fetching static NAT for VM: %s" % self.virtual_machine.name)
ip_addrs = PublicIPAddress.list(
- self.api_client,
- associatednetworkid=self.guest_network.id,
- isstaticnat=True,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ associatednetworkid=self.guest_network.id,
+ isstaticnat=True,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(ip_addrs, list),
- True,
- "List Public IP address should return valid IP address for network"
- )
+ isinstance(ip_addrs, list),
+ True,
+ "List Public IP address should return valid IP address for network"
+ )
static_nat = ip_addrs[0]
self.debug("Static NAT for VM: %s is: %s" % (
- self.virtual_machine.name,
- static_nat.ipaddress
- ))
+ self.virtual_machine.name,
+ static_nat.ipaddress
+ ))
# Fetch details from user_ip_address table in database
self.debug(
- "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \
- % static_nat.ipaddress)
+ "select is_system, one_to_one_nat from user_ip_address\
+ where public_ip_address='%s';" %
+ static_nat.ipaddress)
qresultset = self.dbclient.execute(
- "select is_system, one_to_one_nat from user_ip_address where public_ip_address='%s';" \
- % static_nat.ipaddress
- )
+ "select is_system, one_to_one_nat from user_ip_address\
+ where public_ip_address='%s';" %
+ static_nat.ipaddress)
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "Check DB Query result set"
- )
+ len(qresultset),
+ 0,
+ "Check DB Query result set"
+ )
qresult = qresultset[0]
self.assertEqual(
- qresult[0],
- 1,
- "is_system value should be 1 for automatically assigned IP"
- )
+ qresult[0],
+ 1,
+ "is_system value should be 1 for automatically assigned IP"
+ )
self.assertEqual(
- qresult[1],
- 1,
- "one_to_one_nat value should be 1 for automatically assigned IP"
- )
+ qresult[1],
+ 1,
+ "one_to_one_nat value should be 1 for automatically assigned IP"
+ )
# try:
# self.debug("SSH into VM: %s" % static_nat.ipaddress)
# ssh = self.virtual_machine.get_ssh_client(
@@ -657,48 +672,43 @@ class TestEIP(cloudstackTestCase):
# (static_nat.ipaddress, e))
self.debug("SSH into netscaler: %s" %
- self.services["netscaler"]["ipaddress"])
- try:
- ssh_client = SshClient(
- self.services["netscaler"]["ipaddress"],
- 22,
- self.services["netscaler"]["username"],
- self.services["netscaler"]["password"],
- )
- self.debug("command: show ip")
- res = ssh_client.execute("show ip")
- result = str(res)
- self.debug("Output: %s" % result)
+ self.services["netscaler"]["ipaddress"])
+
+ ssh_client = SshClient(
+ self.services["netscaler"]["ipaddress"],
+ 22,
+ self.services["netscaler"]["username"],
+ self.services["netscaler"]["password"],
+ )
+ self.debug("command: show ip")
+ res = ssh_client.execute("show ip")
+ result = str(res)
+ self.debug("Output: %s" % result)
- self.assertEqual(
- result.count(static_nat.ipaddress),
- 1,
- "One IP from EIP pool should be taken and configured on NS"
- )
+ self.assertEqual(
+ result.count(static_nat.ipaddress),
+ 1,
+ "One IP from EIP pool should be taken and configured on NS"
+ )
- self.debug("Command:show inat")
- res = ssh_client.execute("show inat")
+ self.debug("Command:show inat")
+ res = ssh_client.execute("show inat")
- result = str(res)
- self.debug("Output: %s" % result)
+ result = str(res)
+ self.debug("Output: %s" % result)
- self.assertEqual(
- result.count("USIP: ON"),
- 2,
- "User source IP should be enabled for INAT service"
- )
-
- except Exception as e:
- self.fail("SSH Access failed for %s: %s" % \
- (self.services["netscaler"]["ipaddress"], e))
+ self.assertEqual(
+ result.count("USIP: ON"),
+ 2,
+ "User source IP should be enabled for INAT service"
+ )
return
- @attr(tags = ["eip"])
+ @attr(tags=["eip"])
def test_04_disable_static_nat_system(self):
"""Test disable static NAT with system = True
"""
-
# Validate the following
# 1. Try to disassociate/disable static NAT on EIP where is_system=1
# 2. This operation should fail with proper error message.
@@ -706,53 +716,56 @@ class TestEIP(cloudstackTestCase):
self.debug(
"Fetching static NAT for VM: %s" % self.virtual_machine.name)
ip_addrs = PublicIPAddress.list(
- self.api_client,
- associatednetworkid=self.guest_network.id,
- isstaticnat=True,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ associatednetworkid=self.guest_network.id,
+ isstaticnat=True,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(ip_addrs, list),
- True,
- "List Public IP address should return valid IP address for network"
- )
+ isinstance(ip_addrs, list),
+ True,
+ "List Public IP address should return valid IP address for network"
+ )
static_nat = ip_addrs[0]
self.debug("Static NAT for VM: %s is: %s" % (
- self.virtual_machine.name,
- static_nat.ipaddress
- ))
+ self.virtual_machine.name,
+ static_nat.ipaddress
+ ))
# Fetch details from user_ip_address table in database
self.debug(
- "select is_system from user_ip_address where public_ip_address='%s';" \
- % static_nat.ipaddress)
+ "select is_system from user_ip_address where\
+ public_ip_address='%s';" %
+ static_nat.ipaddress)
qresultset = self.dbclient.execute(
- "select is_system from user_ip_address where public_ip_address='%s';" \
- % static_nat.ipaddress
- )
+ "select is_system from user_ip_address where\
+ public_ip_address='%s';" %
+ static_nat.ipaddress)
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "Check DB Query result set"
- )
+ len(qresultset),
+ 0,
+ "Check DB Query result set"
+ )
qresult = qresultset[0]
self.assertEqual(
- qresult[0],
- 1,
- "user_ip_address.is_system value should be 1"
- )
+ qresult[0],
+ 1,
+ "user_ip_address.is_system value should be 1"
+ )
- self.debug("Disassociate Static NAT: %s" % static_nat.ipaddress)
+ self.debug(
+ "Disassociate Static NAT: %s" %
+ static_nat.ipaddress)
with self.assertRaises(Exception):
cmd = disassociateIpAddress.disassociateIpAddressCmd()
@@ -762,12 +775,11 @@ class TestEIP(cloudstackTestCase):
self.debug("Disassociate system IP failed")
return
- @attr(tags = ["eip"])
+ @attr(tags=["eip"])
def test_05_destroy_instance(self):
"""Test EIO after destroying instance
"""
-
# Validate the following
# 1. Destroy instance. Destroy should result in is_system=0 for EIP
# and EIP should also be marked as free.
@@ -778,104 +790,102 @@ class TestEIP(cloudstackTestCase):
self.debug(
"Fetching static NAT for VM: %s" % self.virtual_machine.name)
ip_addrs = PublicIPAddress.list(
- self.api_client,
- associatednetworkid=self.guest_network.id,
- isstaticnat=True,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ associatednetworkid=self.guest_network.id,
+ isstaticnat=True,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(ip_addrs, list),
- True,
- "List Public IP address should return valid IP address for network"
- )
+ isinstance(ip_addrs, list),
+ True,
+ "List Public IP address should return valid IP address for network"
+ )
static_nat = ip_addrs[0]
self.debug("Static NAT for VM: %s is: %s" % (
- self.virtual_machine.name,
- static_nat.ipaddress
- ))
+ self.virtual_machine.name,
+ static_nat.ipaddress
+ ))
self.debug("Destroying an instance: %s" % self.virtual_machine.name)
self.virtual_machine.delete(self.apiclient, expunge=True)
self.debug("Destroy instance complete!")
vms = VirtualMachine.list(
- self.apiclient,
- id=self.virtual_machine.id
- )
+ self.apiclient,
+ id=self.virtual_machine.id
+ )
self.assertEqual(
- vms,
- None,
- "list VM should not return anything after destroy"
- )
+ vms,
+ None,
+ "list VM should not return anything after destroy"
+ )
# Fetch details from user_ip_address table in database
self.debug(
- "select is_system, state from user_ip_address where public_ip_address='%s';" \
- % static_nat.ipaddress)
+ "select is_system, state from user_ip_address where\
+ public_ip_address='%s';" %
+ static_nat.ipaddress)
qresultset = self.dbclient.execute(
- "select is_system, state from user_ip_address where public_ip_address='%s';" \
- % static_nat.ipaddress
- )
+ "select is_system, state from user_ip_address where\
+ public_ip_address='%s';" %
+ static_nat.ipaddress)
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "Check DB Query result set"
- )
+ len(qresultset),
+ 0,
+ "Check DB Query result set"
+ )
qresult = qresultset[0]
self.assertEqual(
- qresult[0],
- 0,
- "user_ip_address.is_system value should be 0"
- )
+ qresult[0],
+ 0,
+ "user_ip_address.is_system value should be 0"
+ )
self.assertEqual(
- qresult[1],
- "Free",
- "IP should be marked as Free after destroying VM"
- )
+ qresult[1],
+ "Free",
+ "IP should be marked as Free after destroying VM"
+ )
self.debug("SSH into netscaler: %s" %
- self.services["netscaler"]["ipaddress"])
- try:
- ssh_client = SshClient(
- self.services["netscaler"]["ipaddress"],
- 22,
- self.services["netscaler"]["username"],
- self.services["netscaler"]["password"],
- )
- self.debug("command: show ip")
- res = ssh_client.execute("show ip")
- result = str(res)
- self.debug("Output: %s" % result)
+ self.services["netscaler"]["ipaddress"])
+
+ ssh_client = SshClient(
+ self.services["netscaler"]["ipaddress"],
+ 22,
+ self.services["netscaler"]["username"],
+ self.services["netscaler"]["password"],
+ )
+ self.debug("command: show ip")
+ res = ssh_client.execute("show ip")
+ result = str(res)
+ self.debug("Output: %s" % result)
- self.assertEqual(
- result.count(static_nat.ipaddress),
- 0,
- "show ip should return nothing after VM destroy"
- )
+ self.assertEqual(
+ result.count(static_nat.ipaddress),
+ 0,
+ "show ip should return nothing after VM destroy"
+ )
- self.debug("Command:show inat")
- res = ssh_client.execute("show inat")
+ self.debug("Command:show inat")
+ res = ssh_client.execute("show inat")
- result = str(res)
- self.debug("Output: %s" % result)
+ result = str(res)
+ self.debug("Output: %s" % result)
- self.assertEqual(
- result.count(static_nat.ipaddress),
- 0,
- "show inat should return nothing after VM destroy"
- )
- except Exception as e:
- self.fail("SSH Access failed for %s: %s" % \
- (self.services["netscaler"]["ipaddress"], e))
+ self.assertEqual(
+ result.count(static_nat.ipaddress),
+ 0,
+ "show inat should return nothing after VM destroy"
+ )
return
@@ -891,70 +901,81 @@ class TestELB(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
+
+ try:
+ cls.services["netscaler"] = cls.config.__dict__[
+ "netscalerDevice"].__dict__
+ except KeyError:
+ raise unittest.SkipTest("Please make sure you have included netscalerDevice\
+ dict in your config file (keys - ipaddress, username,\
+ password")
+ except Exception as e:
+ raise unittest.SkipTest(e)
+
cls.template = get_template(
- cls.api_client,
- cls.zone.id,
- cls.services["ostype"]
- )
+ 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.api_client,
+ cls.services["service_offering"]
+ )
cls.account = Account.create(
- cls.api_client,
- cls.services["account"],
- admin=True,
- domainid=cls.domain.id
- )
+ cls.api_client,
+ cls.services["account"],
+ admin=True,
+ domainid=cls.domain.id
+ )
# Spawn an instance
cls.vm_1 = VirtualMachine.create(
- cls.api_client,
- cls.services["virtual_machine"],
- accountid=cls.account.name,
- domainid=cls.account.domainid,
- serviceofferingid=cls.service_offering.id
- )
+ cls.api_client,
+ cls.services["virtual_machine"],
+ accountid=cls.account.name,
+ domainid=cls.account.domainid,
+ serviceofferingid=cls.service_offering.id
+ )
cls.vm_2 = VirtualMachine.create(
- cls.api_client,
- cls.services["virtual_machine"],
- accountid=cls.account.name,
- domainid=cls.account.domainid,
- serviceofferingid=cls.service_offering.id
- )
+ cls.api_client,
+ cls.services["virtual_machine"],
+ accountid=cls.account.name,
+ domainid=cls.account.domainid,
+ serviceofferingid=cls.service_offering.id
+ )
networks = Network.list(
- cls.api_client,
- zoneid=cls.zone.id,
- listall=True
- )
+ cls.api_client,
+ zoneid=cls.zone.id,
+ listall=True
+ )
if isinstance(networks, list):
# Basic zone has only one network i.e Basic network
cls.guest_network = networks[0]
else:
raise Exception(
"List networks returned empty response for zone: %s" %
- cls.zone.id)
+ cls.zone.id)
cls.lb_rule = LoadBalancerRule.create(
- cls.api_client,
- cls.services["lbrule"],
- accountid=cls.account.name,
- networkid=cls.guest_network.id,
- domainid=cls.account.domainid
- )
+ cls.api_client,
+ cls.services["lbrule"],
+ accountid=cls.account.name,
+ networkid=cls.guest_network.id,
+ domainid=cls.account.domainid
+ )
cls.lb_rule.assign(cls.api_client, [cls.vm_1, cls.vm_2])
cls._cleanup = [
- cls.account,
- cls.service_offering,
- ]
+ cls.account,
+ cls.service_offering,
+ ]
return
@classmethod
def tearDownClass(cls):
try:
- #Cleanup resources used
+ # Cleanup resources used
cleanup_resources(cls.api_client, cls._cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
@@ -968,18 +989,17 @@ class TestELB(cloudstackTestCase):
def tearDown(self):
try:
- #Clean up, terminate the created network offerings
+ # Clean up, terminate the created network offerings
cleanup_resources(self.apiclient, self.cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
-
- @attr(tags = ["eip"])
+
+ @attr(tags=["eip"])
def test_01_elb_create(self):
"""Test ELB by creating a LB rule
"""
-
# Validate the following
# 1. Deploy 2 instances
# 2. Create LB rule to port 22 for the VMs and try to access VMs with
@@ -994,22 +1014,22 @@ class TestELB(cloudstackTestCase):
# Verify listSecurity groups response
security_groups = SecurityGroup.list(
- self.apiclient,
- account=self.account.name,
- domainid=self.account.domainid
- )
+ self.apiclient,
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
self.assertEqual(
- isinstance(security_groups, list),
- True,
- "Check for list security groups response"
- )
+ isinstance(security_groups, list),
+ True,
+ "Check for list security groups response"
+ )
self.assertEqual(
- len(security_groups),
- 1,
- "Check List Security groups response"
- )
+ len(security_groups),
+ 1,
+ "Check List Security groups response"
+ )
self.debug("List Security groups response: %s" %
- str(security_groups))
+ str(security_groups))
security_group = security_groups[0]
@@ -1029,25 +1049,25 @@ class TestELB(cloudstackTestCase):
self.debug(
"Fetching LB IP for account: %s" % self.account.name)
ip_addrs = PublicIPAddress.list(
- self.api_client,
- associatednetworkid=self.guest_network.id,
- account=self.account.name,
- domainid=self.account.domainid,
- forloadbalancing=True,
- listall=True
- )
+ self.api_client,
+ associatednetworkid=self.guest_network.id,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ forloadbalancing=True,
+ listall=True
+ )
self.assertEqual(
- isinstance(ip_addrs, list),
- True,
- "List Public IP address should return valid IP address for network"
- )
+ isinstance(ip_addrs, list),
+ True,
+ "List Public IP address should return valid IP address for network"
+ )
lb_ip = ip_addrs[0]
self.debug("LB IP generated for account: %s is: %s" % (
- self.account.name,
- lb_ip.ipaddress
- ))
-#TODO: uncomment this after ssh issue is resolved
+ self.account.name,
+ lb_ip.ipaddress
+ ))
+# TODO: uncomment this after ssh issue is resolved
# self.debug("SSHing into VMs using ELB IP: %s" % lb_ip.ipaddress)
# try:
# ssh_1 = self.vm_1.get_ssh_client(ipaddress=lb_ip.ipaddress)
@@ -1086,50 +1106,51 @@ class TestELB(cloudstackTestCase):
# Fetch details from user_ip_address table in database
self.debug(
- "select is_system from user_ip_address where public_ip_address='%s';" \
- % lb_ip.ipaddress)
+ "select is_system from user_ip_address where\
+ public_ip_address='%s';" %
+ lb_ip.ipaddress)
qresultset = self.dbclient.execute(
- "select is_system from user_ip_address where public_ip_address='%s';" \
- % lb_ip.ipaddress
- )
+ "select is_system from user_ip_address where\
+ public_ip_address='%s';" %
+ lb_ip.ipaddress)
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "Check DB Query result set"
- )
+ len(qresultset),
+ 0,
+ "Check DB Query result set"
+ )
qresult = qresultset[0]
self.assertEqual(
- qresult[0],
- 1,
- "is_system value should be 1 for system generated LB rule"
- )
+ qresult[0],
+ 1,
+ "is_system value should be 1 for system generated LB rule"
+ )
self.debug("SSH into netscaler: %s" %
- self.services["netscaler"]["ipaddress"])
+ self.services["netscaler"]["ipaddress"])
try:
ssh_client = SshClient(
- self.services["netscaler"]["ipaddress"],
- 22,
- self.services["netscaler"]["username"],
- self.services["netscaler"]["password"],
- )
+ self.services["netscaler"]["ipaddress"],
+ 22,
+ self.services["netscaler"]["username"],
+ self.services["netscaler"]["password"],
+ )
self.debug("command: show ip")
res = ssh_client.execute("show ip")
result = str(res)
self.debug("Output: %s" % result)
self.assertEqual(
- result.count(lb_ip.ipaddress),
- 1,
- "One IP from EIP pool should be taken and configured on NS"
- )
+ result.count(lb_ip.ipaddress),
+ 1,
+ "One IP from EIP pool should be taken and configured on NS"
+ )
self.debug("Command:show lb vserver")
res = ssh_client.execute("show lb vserver")
@@ -1138,22 +1159,23 @@ class TestELB(cloudstackTestCase):
self.debug("Output: %s" % result)
self.assertEqual(
- result.count("Cloud-VirtualServer-%s-22 (%s:22) - TCP" % (lb_ip.ipaddress, lb_ip.ipaddress)),
- 1,
- "User subnet IP should be enabled for LB service"
- )
+ result.count(
+ "Cloud-VirtualServer-%s-22 (%s:22) - TCP" %
+ (lb_ip.ipaddress,
+ lb_ip.ipaddress)),
+ 1,
+ "User subnet IP should be enabled for LB service")
except Exception as e:
- self.fail("SSH Access failed for %s: %s" % \
+ self.fail("SSH Access failed for %s: %s" %
(self.services["netscaler"]["ipaddress"], e))
return
-
- @attr(tags = ["eip"])
+
+ @attr(tags=["eip"])
def test_02_elb_acquire_and_create(self):
"""Test ELB by acquiring IP and then creating a LB rule
"""
-
# Validate the following
# 1. Deploy 2 instances
# 2. Create LB rule to port 22 for the VMs and try to access VMs with
@@ -1169,31 +1191,31 @@ class TestELB(cloudstackTestCase):
self.debug("Acquiring new IP for network: %s" % self.guest_network.id)
public_ip = PublicIPAddress.create(
- self.apiclient,
- accountid=self.account.name,
- zoneid=self.zone.id,
- domainid=self.account.domainid,
- services=self.services["virtual_machine"]
- )
+ self.apiclient,
+ accountid=self.account.name,
+ zoneid=self.zone.id,
+ domainid=self.account.domainid,
+ services=self.services["virtual_machine"]
+ )
self.debug("IP address: %s is acquired by network: %s" % (
- public_ip.ipaddress,
- self.guest_network.id))
+ public_ip.ipaddress.ipaddress,
+ self.guest_network.id))
self.debug("Creating LB rule for public IP: %s" %
- public_ip.ipaddress)
+ public_ip.ipaddress.ipaddress)
lb_rule = LoadBalancerRule.create(
- self.apiclient,
- self.services["lbrule"],
- accountid=self.account.name,
- ipaddressid=public_ip.ipaddress.id,
- networkid=self.guest_network.id,
- domainid=self.account.domaind
- )
+ self.apiclient,
+ self.services["lbrule"],
+ accountid=self.account.name,
+ ipaddressid=public_ip.ipaddress.id,
+ networkid=self.guest_network.id,
+ domainid=self.account.domainid
+ )
self.debug("Assigning VMs (%s, %s) to LB rule: %s" % (self.vm_1.name,
self.vm_2.name,
lb_rule.name))
lb_rule.assign(self.apiclient, [self.vm_1, self.vm_2])
-#TODO: workaround : add route in the guest VM for SNIP
+# TODO: workaround : add route in the guest VM for SNIP
#
# self.debug("SSHing into VMs using ELB IP: %s" %
# public_ip.ipaddress)
@@ -1233,52 +1255,54 @@ class TestELB(cloudstackTestCase):
# self.fail(
# "SSH Access failed for %s: %s" % (self.vm_1.ipaddress, e))
#
-## Fetch details from user_ip_address table in database
+# Fetch details from user_ip_address table in database
self.debug(
- "select is_system from user_ip_address where public_ip_address='%s';" \
- % public_ip.ipaddress)
+ "select is_system from user_ip_address where\
+ public_ip_address='%s';" %
+ public_ip.ipaddress.ipaddress)
qresultset = self.dbclient.execute(
- "select is_system from user_ip_address where public_ip_address='%s';" \
- % public_ip.ipaddress)
+ "select is_system from user_ip_address where\
+ public_ip_address='%s';" %
+ public_ip.ipaddress.ipaddress)
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "Check DB Query result set"
- )
+ len(qresultset),
+ 0,
+ "Check DB Query result set"
+ )
qresult = qresultset[0]
self.assertEqual(
- qresult[0],
- 0,
- "is_system value should be 0 for non-system generated LB rule"
- )
+ qresult[0],
+ 0,
+ "is_system value should be 0 for non-system generated LB rule"
+ )
self.debug("SSH into netscaler: %s" %
- self.services["netscaler"]["ipaddress"])
+ self.services["netscaler"]["ipaddress"])
try:
ssh_client = SshClient(
- self.services["netscaler"]["ipaddress"],
- 22,
- self.services["netscaler"]["username"],
- self.services["netscaler"]["password"],
- )
+ self.services["netscaler"]["ipaddress"],
+ 22,
+ self.services["netscaler"]["username"],
+ self.services["netscaler"]["password"],
+ )
self.debug("command: show ip")
res = ssh_client.execute("show ip")
result = str(res)
self.debug("Output: %s" % result)
self.assertEqual(
- result.count(public_ip.ipaddress),
- 1,
- "One IP from EIP pool should be taken and configured on NS"
- )
+ result.count(public_ip.ipaddress.ipaddress),
+ 1,
+ "One IP from EIP pool should be taken and configured on NS"
+ )
self.debug("Command:show lb vserver")
res = ssh_client.execute("show lb vserver")
@@ -1287,23 +1311,23 @@ class TestELB(cloudstackTestCase):
self.debug("Output: %s" % result)
self.assertEqual(
- result.count("Cloud-VirtualServer-%s-22 (%s:22) - TCP" % (public_ip.ipaddress, public_ip.ipaddress)),
- 1,
- "User subnet IP should be enabled for LB service"
- )
+ result.count(
+ "Cloud-VirtualServer-%s-22 (%s:22) - TCP" %
+ (public_ip.ipaddress.ipaddress,
+ public_ip.ipaddress.ipaddress)),
+ 1,
+ "User subnet IP should be enabled for LB service")
except Exception as e:
- self.fail("SSH Access failed for %s: %s" % \
+ self.fail("SSH Access failed for %s: %s" %
(self.services["netscaler"]["ipaddress"], e))
return
-
- @attr(tags = ["eip"])
+ @attr(tags=["eip"])
def test_03_elb_delete_lb_system(self):
"""Test delete LB rule generated with public IP with is_system = 1
"""
-
# Validate the following
# 1. Deleting LB rule should release EIP where is_system=1
# 2. check configuration changes for EIP reflects on NS
@@ -1315,73 +1339,69 @@ class TestELB(cloudstackTestCase):
self.debug(
"Fetching LB IP for account: %s" % self.account.name)
ip_addrs = PublicIPAddress.list(
- self.api_client,
- associatednetworkid=self.guest_network.id,
- account=self.account.name,
- domainid=self.account.domainid,
- forloadbalancing=True,
- listall=True
- )
+ self.api_client,
+ associatednetworkid=self.guest_network.id,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ forloadbalancing=True,
+ listall=True
+ )
self.assertEqual(
- isinstance(ip_addrs, list),
- True,
- "List Public IP address should return valid IP address for network"
- )
+ isinstance(ip_addrs, list),
+ True,
+ "List Public IP address should return valid IP address for network"
+ )
lb_ip = ip_addrs[0]
self.debug("LB IP generated for account: %s is: %s" % (
- self.account.name,
- lb_ip.ipaddress
- ))
+ self.account.name,
+ lb_ip.ipaddress
+ ))
self.debug("Deleting LB rule: %s" % self.lb_rule.id)
self.lb_rule.delete(self.apiclient)
-
time.sleep(60)
self.debug("SSH into netscaler: %s" %
- self.services["netscaler"]["ipaddress"])
- try:
- ssh_client = SshClient(
- self.services["netscaler"]["ipaddress"],
- 22,
- self.services["netscaler"]["username"],
- self.services["netscaler"]["password"],
- )
- self.debug("command: show ip")
- res = ssh_client.execute("show ip")
- result = str(res)
- self.debug("Output: %s" % result)
+ self.services["netscaler"]["ipaddress"])
+
+ ssh_client = SshClient(
+ self.services["netscaler"]["ipaddress"],
+ 22,
+ self.services["netscaler"]["username"],
+ self.services["netscaler"]["password"],
+ )
+ self.debug("command: show ip")
+ res = ssh_client.execute("show ip")
+ result = str(res)
+ self.debug("Output: %s" % result)
- self.assertEqual(
- result.count(lb_ip.ipaddress),
- 0,
- "One IP from EIP pool should be taken and configured on NS"
- )
+ self.assertEqual(
+ result.count(lb_ip.ipaddress),
+ 1,
+ "One IP from EIP pool should be taken and configured on NS"
+ )
- self.debug("Command:show lb vserver")
- res = ssh_client.execute("show lb vserver")
+ self.debug("Command:show lb vserver")
+ res = ssh_client.execute("show lb vserver")
- result = str(res)
- self.debug("Output: %s" % result)
-
- self.assertEqual(
- result.count("Cloud-VirtualServer-%s-22 (%s:22) - TCP" % (lb_ip.ipaddress, lb_ip.ipaddress) ),
- 0,
- "User subnet IP should be enabled for LB service"
- )
+ result = str(res)
+ self.debug("Output: %s" % result)
- except Exception as e:
- self.fail("SSH Access failed for %s: %s" % \
- (self.services["netscaler"]["ipaddress"], e))
+ self.assertEqual(
+ result.count(
+ "Cloud-VirtualServer-%s-22 (%s:22) - TCP" %
+ (lb_ip.ipaddress,
+ lb_ip.ipaddress)),
+ 1,
+ "User subnet IP should be enabled for LB service")
return
-
- @attr(tags = ["eip"])
+
+ @attr(tags=["eip"])
def test_04_delete_lb_on_eip(self):
"""Test delete LB rule generated on EIP
"""
-
# Validate the following
# 1. Deleting LB rule won't release EIP where is_system=0
# 2. disassociateIP must release the above IP
@@ -1391,77 +1411,79 @@ class TestELB(cloudstackTestCase):
# * "show lb vserer"-make sure that output says they are all up and
# running and USNIP : ON
- # Fetch details from account_id table in database
+ # Fetch details from account_id table in database
self.debug(
- "select id from account where account_name='%s';" \
- % self.account.name)
+ "select id from account where account_name='%s';"
+ % self.account.name)
qresultset = self.dbclient.execute(
- "select id from account where account_name='%s';" \
- % self.account.name)
+ "select id from account where account_name='%s';"
+ % self.account.name)
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "DB query should return a valid public IP address"
- )
+ len(qresultset),
+ 0,
+ "DB query should return a valid public IP address"
+ )
qresult = qresultset[0]
account_id = qresult[0]
# Fetch details from user_ip_address table in database
self.debug(
- "select public_ip_address from user_ip_address where is_system=0 and account_id=%s;" \
- % account_id)
+ "select public_ip_address from user_ip_address where\
+ is_system=0 and account_id=%s;" %
+ account_id)
qresultset = self.dbclient.execute(
- "select public_ip_address from user_ip_address where is_system=0 and account_id=%s;" \
- % account_id)
+ "select public_ip_address from user_ip_address where\
+ is_system=0 and account_id=%s;" %
+ account_id)
self.assertEqual(
- isinstance(qresultset, list),
- True,
- "Check DB query result set for valid data"
- )
+ isinstance(qresultset, list),
+ True,
+ "Check DB query result set for valid data"
+ )
self.assertNotEqual(
- len(qresultset),
- 0,
- "DB query should return a valid public IP address"
- )
+ len(qresultset),
+ 0,
+ "DB query should return a valid public IP address"
+ )
qresult = qresultset[0]
public_ip = qresult[0]
self.debug(
"Fetching public IP for account: %s" % self.account.name)
ip_addrs = PublicIPAddress.list(
- self.api_client,
- ipaddress=public_ip,
- listall=True
- )
+ self.api_client,
+ ipaddress=public_ip,
+ listall=True
+ )
self.debug("ip address list: %s" % ip_addrs)
self.assertEqual(
- isinstance(ip_addrs, list),
- True,
- "List Public IP address should return valid IP address for network"
- )
+ isinstance(ip_addrs, list),
+ True,
+ "List Public IP address should return valid IP address for network"
+ )
lb_ip = ip_addrs[0]
lb_rules = LoadBalancerRule.list(
- self.apiclient,
- publicipid=lb_ip.id,
- listall=True
- )
+ self.apiclient,
+ publicipid=lb_ip.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(lb_rules, list),
- True,
- "Atleast one LB rule must be present for public IP address"
- )
+ isinstance(lb_rules, list),
+ True,
+ "Atleast one LB rule must be present for public IP address"
+ )
lb_rule = lb_rules[0]
self.debug("Deleting LB rule associated with IP: %s" % public_ip)
@@ -1472,38 +1494,38 @@ class TestELB(cloudstackTestCase):
except Exception as e:
self.fail("Deleting LB rule failed for IP: %s-%s" % (public_ip, e))
-#TODO:check the lb rule list and then confirm that lb rule is deleted
+# TODO:check the lb rule list and then confirm that lb rule is deleted
self.debug("LB rule deleted!")
ip_addrs = PublicIPAddress.list(
- self.api_client,
- ipaddress=public_ip,
- listall=True
- )
+ self.api_client,
+ ipaddress=public_ip,
+ listall=True
+ )
self.assertEqual(
- isinstance(ip_addrs, list),
- True,
- "Deleting LB rule should not delete public IP"
- )
+ isinstance(ip_addrs, list),
+ True,
+ "Deleting LB rule should not delete public IP"
+ )
self.debug("SSH into netscaler: %s" %
- self.services["netscaler"]["ipaddress"])
+ self.services["netscaler"]["ipaddress"])
try:
ssh_client = SshClient(
- self.services["netscaler"]["ipaddress"],
- 22,
- self.services["netscaler"]["username"],
- self.services["netscaler"]["password"],
- )
+ self.services["netscaler"]["ipaddress"],
+ 22,
+ self.services["netscaler"]["username"],
+ self.services["netscaler"]["password"],
+ )
self.debug("command: show ip")
res = ssh_client.execute("show ip")
result = str(res)
self.debug("Output: %s" % result)
self.assertNotEqual(
- result.count(public_ip),
- 1,
- "One IP from EIP pool should be taken and configured on NS"
- )
+ result.count(public_ip),
+ 1,
+ "One IP from EIP pool should be taken and configured on NS"
+ )
self.debug("Command:show lb vserver")
res = ssh_client.execute("show lb vserver")
@@ -1512,12 +1534,14 @@ class TestELB(cloudstackTestCase):
self.debug("Output: %s" % result)
self.assertNotEqual(
- result.count("Cloud-VirtualServer-%s-22 (%s:22) - TCP" % (lb_ip.ipaddress, lb_ip.ipaddress)),
- 1,
- "User subnet IP should be enabled for LB service"
- )
+ result.count(
+ "Cloud-VirtualServer-%s-22 (%s:22) - TCP" %
+ (lb_ip.ipaddress,
+ lb_ip.ipaddress)),
+ 1,
+ "User subnet IP should be enabled for LB service")
except Exception as e:
- self.fail("SSH Access failed for %s: %s" % \
+ self.fail("SSH Access failed for %s: %s" %
(self.services["netscaler"]["ipaddress"], e))
return