apache/cloudstack
1
0
Fork 0
mirror of https://github.com/apache/cloudstack.git synced 2025-12-16 10:32:34 +01:00
Code Issues Packages Projects Releases Wiki Activity

Unwrap the test docstring for the testrunner

Browse Source 
Signed-off-by: Prasanna Santhanam <tsp@apache.org>
This commit is contained in:
Prasanna Santhanam 2013-05-24 14:22:47 +05:30
parent b2be0dbf6b
commit a6d13cb799
1 changed files with 110 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

222
test/integration/smoke/test_network.py
View File

@ -189,8 +189,7 @@ class TestPublicIP(cloudstackTestCase):
@attr(tags = ["advanced", "advancedns", "smoke"]) @attr(tags = ["advanced", "advancedns", "smoke"])
def test_public_ip_admin_account(self): def test_public_ip_admin_account(self):
"""Test for Associate/Disassociate """Test for Associate/Disassociate public IP address for admin account"""
public IP address for admin account"""
# Validate the following: # Validate the following:
# 1. listPubliIpAddresses API returns the list of acquired addresses # 1. listPubliIpAddresses API returns the list of acquired addresses
@ -240,8 +239,7 @@ class TestPublicIP(cloudstackTestCase):
@attr(tags = ["advanced", "advancedns", "smoke"]) @attr(tags = ["advanced", "advancedns", "smoke"])
def test_public_ip_user_account(self): def test_public_ip_user_account(self):
"""Test for Associate/Disassociate """Test for Associate/Disassociate public IP address for user account"""
public IP address for user account"""
# Validate the following: # Validate the following:
# 1. listPubliIpAddresses API returns the list of acquired addresses # 1. listPubliIpAddresses API returns the list of acquired addresses
@ -885,102 +883,102 @@ class TestLoadBalancingRule(cloudstackTestCase):
# Check if VM is in Running state before creating LB rule # Check if VM is in Running state before creating LB rule
vm_response = VirtualMachine.list( vm_response = VirtualMachine.list(
self.apiclient, self.apiclient,
account=self.account.name, account=self.account.name,
domainid=self.account.domainid domainid=self.account.domainid
) )
self.assertEqual( self.assertEqual(
isinstance(vm_response, list), isinstance(vm_response, list),
True, True,
"Check list VM returns a valid list" "Check list VM returns a valid list"
) )
self.assertNotEqual( self.assertNotEqual(
len(vm_response), len(vm_response),
0, 0,
"Check Port Forwarding Rule is created" "Check Port Forwarding Rule is created"
) )
for vm in vm_response: for vm in vm_response:
self.assertEqual( self.assertEqual(
vm.state, vm.state,
'Running', 'Running',
"VM state should be Running before creating a NAT rule." "VM state should be Running before creating a NAT rule."
) )
#Create Load Balancer rule and assign VMs to rule #Create Load Balancer rule and assign VMs to rule
lb_rule = LoadBalancerRule.create( lb_rule = LoadBalancerRule.create(
self.apiclient, self.apiclient,
self.services["lbrule"], self.services["lbrule"],
self.non_src_nat_ip.ipaddress.id, self.non_src_nat_ip.ipaddress.id,
accountid=self.account.name accountid=self.account.name
) )
self.cleanup.append(lb_rule) self.cleanup.append(lb_rule)
lb_rule.assign(self.apiclient, [self.vm_1, self.vm_2]) lb_rule.assign(self.apiclient, [self.vm_1, self.vm_2])
lb_rules = list_lb_rules( lb_rules = list_lb_rules(
self.apiclient, self.apiclient,
id=lb_rule.id id=lb_rule.id
) )
self.assertEqual( self.assertEqual(
isinstance(lb_rules, list), isinstance(lb_rules, list),
True, True,
"Check list response returns a valid list" "Check list response returns a valid list"
) )
#verify listLoadBalancerRules lists the added load balancing rule #verify listLoadBalancerRules lists the added load balancing rule
self.assertNotEqual( self.assertNotEqual(
len(lb_rules), len(lb_rules),
0, 0,
"Check Load Balancer Rule in its List" "Check Load Balancer Rule in its List"
) )
self.assertEqual( self.assertEqual(
lb_rules[0].id, lb_rules[0].id,
lb_rule.id, lb_rule.id,
"Check List Load Balancer Rules returns valid Rule" "Check List Load Balancer Rules returns valid Rule"
) )
# listLoadBalancerRuleInstances should list # listLoadBalancerRuleInstances should list
# all instances associated with that LB rule # all instances associated with that LB rule
lb_instance_rules = list_lb_instances( lb_instance_rules = list_lb_instances(
self.apiclient, self.apiclient,
id=lb_rule.id id=lb_rule.id
) )
self.assertEqual( self.assertEqual(
isinstance(lb_instance_rules, list), isinstance(lb_instance_rules, list),
True, True,
"Check list response returns a valid list" "Check list response returns a valid list"
) )
self.assertNotEqual( self.assertNotEqual(
len(lb_instance_rules), len(lb_instance_rules),
0, 0,
"Check Load Balancer instances Rule in its List" "Check Load Balancer instances Rule in its List"
) )
self.assertIn( self.assertIn(
lb_instance_rules[0].id, lb_instance_rules[0].id,
[self.vm_1.id, self.vm_2.id], [self.vm_1.id, self.vm_2.id],
"Check List Load Balancer instances Rules returns valid VM ID" "Check List Load Balancer instances Rules returns valid VM ID"
) )
self.assertIn( self.assertIn(
lb_instance_rules[1].id, lb_instance_rules[1].id,
[self.vm_1.id, self.vm_2.id], [self.vm_1.id, self.vm_2.id],
"Check List Load Balancer instances Rules returns valid VM ID" "Check List Load Balancer instances Rules returns valid VM ID"
) )
try: try:
self.debug("SSHing into IP address: %s after adding VMs (ID: %s , %s)" % self.debug("SSHing into IP address: %s after adding VMs (ID: %s , %s)" %
( (
self.non_src_nat_ip.ipaddress.ipaddress, self.non_src_nat_ip.ipaddress.ipaddress,
self.vm_1.id, self.vm_1.id,
self.vm_2.id self.vm_2.id
)) ))
ssh_1 = remoteSSHClient( ssh_1 = remoteSSHClient(
self.non_src_nat_ip.ipaddress.ipaddress, self.non_src_nat_ip.ipaddress.ipaddress,
self.services['lbrule']["publicport"], self.services['lbrule']["publicport"],
self.vm_1.username, self.vm_1.username,
self.vm_1.password self.vm_1.password
) )
# If Round Robin Algorithm is chosen, # If Round Robin Algorithm is chosen,
# each ssh command should alternate between VMs # each ssh command should alternate between VMs
@ -989,71 +987,71 @@ class TestLoadBalancingRule(cloudstackTestCase):
time.sleep(self.services["lb_switch_wait"]) time.sleep(self.services["lb_switch_wait"])
self.debug("SSHing again into IP address: %s with VMs (ID: %s , %s) added to LB rule" % self.debug("SSHing again into IP address: %s with VMs (ID: %s , %s) added to LB rule" %
( (
self.non_src_nat_ip.ipaddress.ipaddress, self.non_src_nat_ip.ipaddress.ipaddress,
self.vm_1.id, self.vm_1.id,
self.vm_2.id self.vm_2.id
)) ))
ssh_2 = remoteSSHClient( ssh_2 = remoteSSHClient(
self.non_src_nat_ip.ipaddress.ipaddress, self.non_src_nat_ip.ipaddress.ipaddress,
self.services['lbrule']["publicport"], self.services['lbrule']["publicport"],
self.vm_1.username, self.vm_1.username,
self.vm_1.password self.vm_1.password
) )
hostnames.append(ssh_2.execute("hostname")[0]) hostnames.append(ssh_2.execute("hostname")[0])
self.debug("Hostnames after adding 2 VMs to LB rule: %s" % str(hostnames)) self.debug("Hostnames after adding 2 VMs to LB rule: %s" % str(hostnames))
self.assertIn( self.assertIn(
self.vm_1.name, self.vm_1.name,
hostnames, hostnames,
"Check if ssh succeeded for server1" "Check if ssh succeeded for server1"
) )
self.assertIn( self.assertIn(
self.vm_2.name, self.vm_2.name,
hostnames, hostnames,
"Check if ssh succeeded for server2" "Check if ssh succeeded for server2"
) )
#SSH should pass till there is a last VM associated with LB rule #SSH should pass till there is a last VM associated with LB rule
lb_rule.remove(self.apiclient, [self.vm_2]) lb_rule.remove(self.apiclient, [self.vm_2])
self.debug("SSHing into IP address: %s after removing VM (ID: %s) from LB rule" % self.debug("SSHing into IP address: %s after removing VM (ID: %s) from LB rule" %
( (
self.non_src_nat_ip.ipaddress.ipaddress, self.non_src_nat_ip.ipaddress.ipaddress,
self.vm_2.id self.vm_2.id
)) ))
ssh_1 = remoteSSHClient( ssh_1 = remoteSSHClient(
self.non_src_nat_ip.ipaddress.ipaddress, self.non_src_nat_ip.ipaddress.ipaddress,
self.services['lbrule']["publicport"], self.services['lbrule']["publicport"],
self.vm_1.username, self.vm_1.username,
self.vm_1.password self.vm_1.password
) )
hostnames.append(ssh_1.execute("hostname")[0]) hostnames.append(ssh_1.execute("hostname")[0])
self.debug("Hostnames after removing VM2: %s" % str(hostnames)) self.debug("Hostnames after removing VM2: %s" % str(hostnames))
except Exception as e: except Exception as e:
self.fail("%s: SSH failed for VM with IP Address: %s" % self.fail("%s: SSH failed for VM with IP Address: %s" %
(e, self.non_src_nat_ip.ipaddress.ipaddress)) (e, self.non_src_nat_ip.ipaddress.ipaddress))
self.assertIn( self.assertIn(
self.vm_1.name, self.vm_1.name,
hostnames, hostnames,
"Check if ssh succeeded for server1" "Check if ssh succeeded for server1"
) )
lb_rule.remove(self.apiclient, [self.vm_1]) lb_rule.remove(self.apiclient, [self.vm_1])
with self.assertRaises(Exception): with self.assertRaises(Exception):
self.fail("SSHing into IP address: %s after removing VM (ID: %s) from LB rule" % self.debug("SSHing into IP address: %s after removing VM (ID: %s) from LB rule" %
( (
self.non_src_nat_ip.ipaddress.ipaddress, self.non_src_nat_ip.ipaddress.ipaddress,
self.vm_1.id self.vm_1.id
)) ))
ssh_1 = remoteSSHClient( ssh_1 = remoteSSHClient(
self.non_src_nat_ip.ipaddress.ipaddress, self.non_src_nat_ip.ipaddress.ipaddress,
self.services['lbrule']["publicport"], self.services['lbrule']["publicport"],
self.vm_1.username, self.vm_1.username,
self.vm_1.password self.vm_1.password
) )
ssh_1.execute("hostname")[0] ssh_1.execute("hostname")[0]
return return
@ -1556,7 +1554,7 @@ class TestReleaseIP(cloudstackTestCase):
@attr(tags = ["advanced", "advancedns", "smoke"]) @attr(tags = ["advanced", "advancedns", "smoke"])
def test_releaseIP(self): def test_releaseIP(self):
"""Test for Associate/Disassociate public IP address""" """Test for release public IP address"""
self.debug("Deleting Public IP : %s" % self.ip_addr.id) self.debug("Deleting Public IP : %s" % self.ip_addr.id)