You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by ts...@apache.org on 2012/09/12 15:24:11 UTC
[28/37] Various fixes to tests
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/b433b1af/test/integration/component/test_security_groups.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_security_groups.py b/test/integration/component/test_security_groups.py
index 71e7a29..85c96d3 100644
--- a/test/integration/component/test_security_groups.py
+++ b/test/integration/component/test_security_groups.py
@@ -37,7 +37,7 @@ class Services:
def __init__(self):
self.services = {
- "disk_offering":{
+ "disk_offering": {
"displaytext": "Small",
"name": "Small",
"disksize": 1
@@ -47,12 +47,12 @@ class Services:
"firstname": "Test",
"lastname": "User",
"username": "test",
- # Random characters are appended in create account to
+ # Random characters are appended in create account to
# ensure unique username generated each time
"password": "fr3sca",
},
"virtual_machine": {
- # Create a small virtual machine instance with disk offering
+ # Create a small virtual machine instance with disk offering
"displayname": "Test VM",
"username": "root", # VM creds for SSH
"password": "password",
@@ -65,15 +65,15 @@ class Services:
},
"host": {
"publicport": 22,
- "username": "root", # Host creds for SSH
+ "username": "root", # Host creds for SSH
"password": "fr3sca",
},
"service_offering": {
"name": "Tiny Instance",
"displaytext": "Tiny Instance",
"cpunumber": 1,
- "cpuspeed": 100, # in MHz
- "memory": 64, # In MBs
+ "cpuspeed": 100, # in MHz
+ "memory": 64, # In MBs
},
"security_group": {
"name": 'SSH',
@@ -87,19 +87,19 @@ class Services:
"protocol": 'ICMP',
"startport": -1,
"endport": -1,
- "cidrlist": '0.0.0.0/0',
+ "cidrlist": '0.0.0.0/0',
},
"ostypeid": '0c2c5d19-525b-41be-a8c3-c6607412f82b',
# CentOS 5.3 (64-bit)
"sleep": 60,
"timeout": 10,
- "mode":'basic',
+ "mode": 'basic',
# Networking mode: Basic or Advanced
}
class TestDefaultSecurityGroup(cloudstackTestCase):
-
+
def setUp(self):
self.apiclient = self.testClient.getApiClient()
@@ -125,7 +125,7 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
# Get Zone, Domain and templates
cls.domain = get_domain(cls.api_client, cls.services)
cls.zone = get_zone(cls.api_client, cls.services)
-
+
template = get_template(
cls.api_client,
cls.zone.id,
@@ -134,7 +134,7 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
cls.services["domainid"] = cls.domain.id
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
cls.services["virtual_machine"]["template"] = template.id
-
+
cls.service_offering = ServiceOffering.create(
cls.api_client,
cls.services["service_offering"]
@@ -168,12 +168,12 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
def test_01_deployVM_InDefaultSecurityGroup(self):
"""Test deploy VM in default security group
"""
-
+
tags = ["sg", "eip"]
# Validate the following:
# 1. deploy Virtual machine using admin user
- # 2. listVM should show a VM in Running state
+ # 2. listVM should show a VM in Running state
# 3. listRouters should show one router running
self.virtual_machine = VirtualMachine.create(
@@ -185,7 +185,7 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
)
self.debug("Deployed VM with ID: %s" % self.virtual_machine.id)
self.cleanup.append(self.virtual_machine)
-
+
list_vm_response = list_virtual_machines(
self.apiclient,
id=self.virtual_machine.id
@@ -219,8 +219,8 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
self.virtual_machine.displayname,
"Check virtual machine displayname in listVirtualMachines"
)
-
- # Verify List Routers response for account
+
+ # Verify List Routers response for account
self.debug(
"Verify list routers response for account: %s" \
% self.account.account.name
@@ -235,11 +235,11 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
True,
"Check for list Routers response"
)
-
+
self.debug("Router Response: %s" % routers)
self.assertEqual(
- len(routers),
- 1,
+ len(routers),
+ 1,
"Check virtual router is created for account or not"
)
return
@@ -247,14 +247,14 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
def test_02_listSecurityGroups(self):
"""Test list security groups for admin account
"""
-
+
tags = ["sg", "eip"]
# Validate the following:
# 1. listSecurityGroups in admin account
# 2. There should be one security group (default) listed for the admin account
# 3. No Ingress Rules should be part of the default security group
-
+
sercurity_groups = SecurityGroup.list(
self.apiclient,
account=self.account.account.name,
@@ -266,11 +266,11 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
"Check for list security groups response"
)
self.assertNotEqual(
- len(sercurity_groups),
- 0,
+ len(sercurity_groups),
+ 0,
"Check List Security groups response"
)
- self.debug("List Security groups response: %s" %
+ self.debug("List Security groups response: %s" %
str(sercurity_groups))
self.assertEqual(
hasattr(sercurity_groups, 'ingressrule'),
@@ -278,16 +278,16 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
"Check ingress rule attribute for default security group"
)
return
-
+
def test_03_accessInDefaultSecurityGroup(self):
"""Test access in default security group
"""
-
+
tags = ["sg", "eip"]
# Validate the following:
# 1. deploy Virtual machine using admin user
- # 2. listVM should show a VM in Running state
+ # 2. listVM should show a VM in Running state
# 3. listRouters should show one router running
self.virtual_machine = VirtualMachine.create(
@@ -299,7 +299,7 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
)
self.debug("Deployed VM with ID: %s" % self.virtual_machine.id)
self.cleanup.append(self.virtual_machine)
-
+
list_vm_response = list_virtual_machines(
self.apiclient,
id=self.virtual_machine.id
@@ -309,7 +309,7 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
True,
"Check for list VM response"
)
-
+
self.debug(
"Verify listVirtualMachines response for virtual machine: %s" \
% self.virtual_machine.id
@@ -345,12 +345,12 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
True,
"Check for list security groups response"
)
-
- self.debug("List Security groups response: %s" %
+
+ self.debug("List Security groups response: %s" %
str(sercurity_groups))
self.assertNotEqual(
- len(sercurity_groups),
- 0,
+ len(sercurity_groups),
+ 0,
"Check List Security groups response"
)
self.assertEqual(
@@ -358,7 +358,7 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
False,
"Check ingress rule attribute for default security group"
)
-
+
# SSH Attempt to VM should fail
with self.assertRaises(Exception):
self.debug("SSH into VM: %s" % self.virtual_machine.ssh_ip)
@@ -369,10 +369,10 @@ class TestDefaultSecurityGroup(cloudstackTestCase):
self.virtual_machine.password
)
return
-
+
class TestAuthorizeIngressRule(cloudstackTestCase):
-
+
def setUp(self):
self.apiclient = self.testClient.getApiClient()
@@ -398,7 +398,7 @@ class TestAuthorizeIngressRule(cloudstackTestCase):
# Get Zone, Domain and templates
cls.domain = get_domain(cls.api_client, cls.services)
cls.zone = get_zone(cls.api_client, cls.services)
-
+
template = get_template(
cls.api_client,
cls.zone.id,
@@ -407,7 +407,7 @@ class TestAuthorizeIngressRule(cloudstackTestCase):
cls.services["domainid"] = cls.domain.id
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
cls.services["virtual_machine"]["template"] = template.id
-
+
cls.service_offering = ServiceOffering.create(
cls.api_client,
cls.services["service_offering"]
@@ -439,7 +439,7 @@ class TestAuthorizeIngressRule(cloudstackTestCase):
def test_01_authorizeIngressRule(self):
"""Test authorize ingress rule
"""
-
+
tags = ["sg", "eip"]
# Validate the following:
@@ -447,11 +447,11 @@ class TestAuthorizeIngressRule(cloudstackTestCase):
#2. Createsecuritygroup (ssh-incoming) for this account
#3. authorizeSecurityGroupIngress to allow ssh access to the VM
#4. deployVirtualMachine into this security group (ssh-incoming)
-
+
security_group = SecurityGroup.create(
- self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.apiclient,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.debug("Created security group with ID: %s" % security_group.id)
@@ -466,17 +466,17 @@ class TestAuthorizeIngressRule(cloudstackTestCase):
True,
"Check for list security groups response"
)
-
+
self.assertEqual(
- len(sercurity_groups),
- 2,
+ len(sercurity_groups),
+ 2,
"Check List Security groups response"
)
# Authorize Security group to SSH to VM
ingress_rule = security_group.authorize(
self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.assertEqual(
@@ -484,8 +484,8 @@ class TestAuthorizeIngressRule(cloudstackTestCase):
True,
"Check ingress rule created properly"
)
-
- self.debug("Authorizing ingress rule for sec group ID: %s for ssh access"
+
+ self.debug("Authorizing ingress rule for sec group ID: %s for ssh access"
% security_group.id)
self.virtual_machine = VirtualMachine.create(
self.apiclient,
@@ -508,7 +508,7 @@ class TestAuthorizeIngressRule(cloudstackTestCase):
class TestRevokeIngressRule(cloudstackTestCase):
-
+
def setUp(self):
self.apiclient = self.testClient.getApiClient()
@@ -534,7 +534,7 @@ class TestRevokeIngressRule(cloudstackTestCase):
# Get Zone, Domain and templates
cls.domain = get_domain(cls.api_client, cls.services)
cls.zone = get_zone(cls.api_client, cls.services)
-
+
template = get_template(
cls.api_client,
cls.zone.id,
@@ -543,7 +543,7 @@ class TestRevokeIngressRule(cloudstackTestCase):
cls.services["domainid"] = cls.domain.id
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
cls.services["virtual_machine"]["template"] = template.id
-
+
cls.service_offering = ServiceOffering.create(
cls.api_client,
cls.services["service_offering"]
@@ -575,7 +575,7 @@ class TestRevokeIngressRule(cloudstackTestCase):
def test_01_revokeIngressRule(self):
"""Test revoke ingress rule
"""
-
+
tags = ["sg", "eip"]
# Validate the following:
@@ -584,15 +584,15 @@ class TestRevokeIngressRule(cloudstackTestCase):
#3. authorizeSecurityGroupIngress to allow ssh access to the VM
#4. deployVirtualMachine into this security group (ssh-incoming)
#5. Revoke the ingress rule, SSH access should fail
-
+
security_group = SecurityGroup.create(
- self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.apiclient,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.debug("Created security group with ID: %s" % security_group.id)
-
+
# Default Security group should not have any ingress rule
sercurity_groups = SecurityGroup.list(
self.apiclient,
@@ -604,28 +604,28 @@ class TestRevokeIngressRule(cloudstackTestCase):
True,
"Check for list security groups response"
)
-
+
self.assertEqual(
- len(sercurity_groups),
- 2,
+ len(sercurity_groups),
+ 2,
"Check List Security groups response"
)
# Authorize Security group to SSH to VM
- self.debug("Authorizing ingress rule for sec group ID: %s for ssh access"
+ self.debug("Authorizing ingress rule for sec group ID: %s for ssh access"
% security_group.id)
ingress_rule = security_group.authorize(
self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
-
+
self.assertEqual(
isinstance(ingress_rule, dict),
True,
"Check ingress rule created properly"
)
-
+
ssh_rule = (ingress_rule["ingressrule"][0]).__dict__
self.virtual_machine = VirtualMachine.create(
self.apiclient,
@@ -636,7 +636,7 @@ class TestRevokeIngressRule(cloudstackTestCase):
securitygroupids=[security_group.id]
)
self.debug("Deploying VM in account: %s" % self.account.account.name)
-
+
# Should be able to SSH VM
try:
self.debug("SSH into VM: %s" % self.virtual_machine.id)
@@ -645,12 +645,12 @@ class TestRevokeIngressRule(cloudstackTestCase):
self.fail("SSH Access failed for %s: %s" % \
(self.virtual_machine.ipaddress, e)
)
-
- self.debug("Revoking ingress rule for sec group ID: %s for ssh access"
+
+ self.debug("Revoking ingress rule for sec group ID: %s for ssh access"
% security_group.id)
# Revoke Security group to SSH to VM
result = security_group.revoke(
- self.apiclient,
+ self.apiclient,
id=ssh_rule["ruleid"]
)
@@ -667,7 +667,7 @@ class TestRevokeIngressRule(cloudstackTestCase):
class TestDhcpOnlyRouter(cloudstackTestCase):
-
+
def setUp(self):
self.apiclient = self.testClient.getApiClient()
@@ -693,17 +693,17 @@ class TestDhcpOnlyRouter(cloudstackTestCase):
# Get Zone, Domain and templates
cls.domain = get_domain(cls.api_client, cls.services)
cls.zone = get_zone(cls.api_client, cls.services)
-
+
template = get_template(
cls.api_client,
cls.zone.id,
cls.services["ostypeid"]
)
-
+
cls.services["domainid"] = cls.domain.id
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
cls.services["virtual_machine"]["template"] = template.id
-
+
cls.service_offering = ServiceOffering.create(
cls.api_client,
cls.services["service_offering"]
@@ -742,13 +742,13 @@ class TestDhcpOnlyRouter(cloudstackTestCase):
def test_01_dhcpOnlyRouter(self):
"""Test router services for user account
"""
-
+
tags = ["sg", "eip", "basic"]
-
+
# Validate the following
#1. List routers for any user account
#2. The only service supported by this router should be dhcp
-
+
# Find router associated with user account
list_router_response = list_routers(
self.apiclient,
@@ -767,7 +767,7 @@ class TestDhcpOnlyRouter(cloudstackTestCase):
zoneid=router.zoneid,
type='Routing',
state='Up',
- virtualmachineid=self.virtual_machine.id
+ id=router.hostid
)
self.assertEqual(
isinstance(hosts, list),
@@ -775,7 +775,7 @@ class TestDhcpOnlyRouter(cloudstackTestCase):
"Check list host returns a valid list"
)
host = hosts[0]
-
+
self.debug("Router ID: %s, state: %s" % (router.id, router.state))
self.assertEqual(
@@ -801,10 +801,10 @@ class TestDhcpOnlyRouter(cloudstackTestCase):
"Check dnsmasq service is running or not"
)
return
-
+
class TestdeployVMWithUserData(cloudstackTestCase):
-
+
def setUp(self):
self.apiclient = self.testClient.getApiClient()
@@ -830,17 +830,17 @@ class TestdeployVMWithUserData(cloudstackTestCase):
# Get Zone, Domain and templates
cls.domain = get_domain(cls.api_client, cls.services)
cls.zone = get_zone(cls.api_client, cls.services)
-
+
template = get_template(
cls.api_client,
cls.zone.id,
cls.services["ostypeid"]
)
-
+
cls.services["domainid"] = cls.domain.id
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
cls.services["virtual_machine"]["template"] = template.id
-
+
cls.service_offering = ServiceOffering.create(
cls.api_client,
cls.services["service_offering"]
@@ -868,12 +868,12 @@ class TestdeployVMWithUserData(cloudstackTestCase):
raise Exception("Warning: Exception during cleanup : %s" % e)
return
-
+
def test_01_deployVMWithUserData(self):
"""Test Deploy VM with User data"""
-
+
tags = ["sg", "eip"]
-
+
# Validate the following
# 1. CreateAccount of type user
# 2. CreateSecurityGroup ssh-incoming
@@ -881,7 +881,7 @@ class TestdeployVMWithUserData(cloudstackTestCase):
# 4. deployVirtualMachine into this group with some base64 encoded user-data
# 5. wget http://10.1.1.1/latest/user-data to get the latest userdata from the
# router for this VM
-
+
# Find router associated with user account
list_router_response = list_routers(
self.apiclient,
@@ -894,11 +894,11 @@ class TestdeployVMWithUserData(cloudstackTestCase):
"Check list response returns a valid list"
)
router = list_router_response[0]
-
+
security_group = SecurityGroup.create(
- self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.apiclient,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.debug("Created security group with ID: %s" % security_group.id)
@@ -914,23 +914,23 @@ class TestdeployVMWithUserData(cloudstackTestCase):
"Check for list security groups response"
)
self.assertEqual(
- len(sercurity_groups),
- 2,
+ len(sercurity_groups),
+ 2,
"Check List Security groups response"
)
-
+
self.debug(
"Authorize Ingress Rule for Security Group %s for account: %s" \
% (
security_group.id,
self.account.account.name
))
-
+
# Authorize Security group to SSH to VM
ingress_rule = security_group.authorize(
self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.assertEqual(
@@ -953,13 +953,13 @@ class TestdeployVMWithUserData(cloudstackTestCase):
"SSH to VM with IP Address: %s"\
% self.virtual_machine.ssh_ip
)
-
+
ssh = self.virtual_machine.get_ssh_client()
except Exception as e:
self.fail("SSH Access failed for %s: %s" % \
(self.virtual_machine.ipaddress, e)
)
-
+
cmds = [
"wget http://%s/latest/user-data" % router.guestipaddress,
"cat user-data",
@@ -967,39 +967,39 @@ class TestdeployVMWithUserData(cloudstackTestCase):
for c in cmds:
result = ssh.execute(c)
self.debug("%s: %s" % (c, result))
-
+
res = str(result)
self.assertEqual(
res.count(self.services["virtual_machine"]["userdata"]),
- 1,
+ 1,
"Verify user data"
)
return
class TestDeleteSecurityGroup(cloudstackTestCase):
-
+
def setUp(self):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
-
+
self.services = Services().services
# Get Zone, Domain and templates
self.domain = get_domain(self.apiclient, self.services)
self.zone = get_zone(self.apiclient, self.services)
-
+
template = get_template(
self.apiclient,
self.zone.id,
self.services["ostypeid"]
)
-
+
self.services["domainid"] = self.domain.id
self.services["virtual_machine"]["zoneid"] = self.zone.id
self.services["virtual_machine"]["template"] = template.id
-
+
self.service_offering = ServiceOffering.create(
self.apiclient,
self.services["service_offering"]
@@ -1044,23 +1044,23 @@ class TestDeleteSecurityGroup(cloudstackTestCase):
raise Exception("Warning: Exception during cleanup : %s" % e)
return
-
+
def test_01_delete_security_grp_running_vm(self):
"""Test delete security group with running VM"""
-
+
tags = ["sg", "eip"]
-
+
# Validate the following
# 1. createsecuritygroup (ssh-incoming) for this account
# 2. authorizeSecurityGroupIngress to allow ssh access to the VM
# 3. deployVirtualMachine into this security group (ssh-incoming)
# 4. deleteSecurityGroup created in step 1. Deletion should fail
# complaining there are running VMs in this group
-
+
security_group = SecurityGroup.create(
- self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.apiclient,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.debug("Created security group with ID: %s" % security_group.id)
@@ -1075,10 +1075,10 @@ class TestDeleteSecurityGroup(cloudstackTestCase):
True,
"Check for list security groups response"
)
-
+
self.assertEqual(
- len(sercurity_groups),
- 2,
+ len(sercurity_groups),
+ 2,
"Check List Security groups response"
)
self.debug(
@@ -1087,12 +1087,12 @@ class TestDeleteSecurityGroup(cloudstackTestCase):
security_group.id,
self.account.account.name
))
-
+
# Authorize Security group to SSH to VM
ingress_rule = security_group.authorize(
self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.assertEqual(
@@ -1100,7 +1100,7 @@ class TestDeleteSecurityGroup(cloudstackTestCase):
True,
"Check ingress rule created properly"
)
-
+
self.virtual_machine = VirtualMachine.create(
self.apiclient,
self.services["virtual_machine"],
@@ -1110,41 +1110,41 @@ class TestDeleteSecurityGroup(cloudstackTestCase):
securitygroupids=[security_group.id]
)
self.debug("Deploying VM in account: %s" % self.account.account.name)
-
+
# Deleting Security group should raise exception
security_group.delete(self.apiclient)
-
+
#sleep to ensure that Security group is deleted properly
time.sleep(self.services["sleep"])
-
+
# Default Security group should not have any ingress rule
sercurity_groups = SecurityGroup.list(
self.apiclient,
id=security_group.id
)
self.assertNotEqual(
- sercurity_groups,
- None,
+ sercurity_groups,
+ None,
"Check List Security groups response"
)
return
def test_02_delete_security_grp_withoout_running_vm(self):
"""Test delete security group without running VM"""
-
+
tags = ["sg", "eip"]
-
+
# Validate the following
# 1. createsecuritygroup (ssh-incoming) for this account
# 2. authorizeSecurityGroupIngress to allow ssh access to the VM
# 3. deployVirtualMachine into this security group (ssh-incoming)
# 4. deleteSecurityGroup created in step 1. Deletion should fail
# complaining there are running VMs in this group
-
+
security_group = SecurityGroup.create(
- self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.apiclient,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.debug("Created security group with ID: %s" % security_group.id)
@@ -1160,11 +1160,11 @@ class TestDeleteSecurityGroup(cloudstackTestCase):
"Check for list security groups response"
)
self.assertEqual(
- len(sercurity_groups),
- 2,
+ len(sercurity_groups),
+ 2,
"Check List Security groups response"
)
-
+
self.debug(
"Authorize Ingress Rule for Security Group %s for account: %s" \
% (
@@ -1174,8 +1174,8 @@ class TestDeleteSecurityGroup(cloudstackTestCase):
# Authorize Security group to SSH to VM
ingress_rule = security_group.authorize(
self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.assertEqual(
@@ -1183,7 +1183,7 @@ class TestDeleteSecurityGroup(cloudstackTestCase):
True,
"Check ingress rule created properly"
)
-
+
self.virtual_machine = VirtualMachine.create(
self.apiclient,
self.services["virtual_machine"],
@@ -1193,10 +1193,10 @@ class TestDeleteSecurityGroup(cloudstackTestCase):
securitygroupids=[security_group.id]
)
self.debug("Deploying VM in account: %s" % self.account.account.name)
-
+
# Destroy the VM
self.virtual_machine.delete(self.apiclient)
-
+
config = list_configurations(
self.apiclient,
name='expunge.delay'
@@ -1210,7 +1210,7 @@ class TestDeleteSecurityGroup(cloudstackTestCase):
self.debug("expunge.delay: %s" % response.value)
# Wait for some time more than expunge.delay
time.sleep(int(response.value) * 2)
-
+
# Deleting Security group should raise exception
try:
self.debug("Deleting Security Group: %s" % security_group.id)
@@ -1220,32 +1220,32 @@ class TestDeleteSecurityGroup(cloudstackTestCase):
% security_group.id
)
return
-
+
class TestIngressRule(cloudstackTestCase):
-
+
def setUp(self):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
-
+
self.services = Services().services
# Get Zone, Domain and templates
self.domain = get_domain(self.apiclient, self.services)
self.zone = get_zone(self.apiclient, self.services)
-
+
template = get_template(
self.apiclient,
self.zone.id,
self.services["ostypeid"]
)
-
+
self.services["domainid"] = self.domain.id
self.services["virtual_machine"]["zoneid"] = self.zone.id
self.services["virtual_machine"]["template"] = template.id
-
+
self.service_offering = ServiceOffering.create(
self.apiclient,
self.services["service_offering"]
@@ -1293,20 +1293,20 @@ class TestIngressRule(cloudstackTestCase):
def test_01_authorizeIngressRule_AfterDeployVM(self):
"""Test delete security group with running VM"""
-
+
tags = ["sg", "eip"]
-
+
# Validate the following
# 1. createsecuritygroup (ssh-incoming, 22via22) for this account
# 2. authorizeSecurityGroupIngress to allow ssh access to the VM
# 3. deployVirtualMachine into this security group (ssh-incoming)
# 4. authorizeSecurityGroupIngress to allow ssh access (startport:222 to
# endport:22) to the VM
-
+
security_group = SecurityGroup.create(
- self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.apiclient,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.debug("Created security group with ID: %s" % security_group.id)
@@ -1322,8 +1322,8 @@ class TestIngressRule(cloudstackTestCase):
"Check for list security groups response"
)
self.assertEqual(
- len(sercurity_groups),
- 2,
+ len(sercurity_groups),
+ 2,
"Check List Security groups response"
)
self.debug(
@@ -1332,12 +1332,12 @@ class TestIngressRule(cloudstackTestCase):
security_group.id,
self.account.account.name
))
-
+
# Authorize Security group to SSH to VM
ingress_rule_1 = security_group.authorize(
self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.assertEqual(
@@ -1354,7 +1354,7 @@ class TestIngressRule(cloudstackTestCase):
securitygroupids=[security_group.id]
)
self.debug("Deploying VM in account: %s" % self.account.account.name)
-
+
self.debug(
"Authorize Ingress Rule for Security Group %s for account: %s" \
% (
@@ -1364,8 +1364,8 @@ class TestIngressRule(cloudstackTestCase):
# Authorize Security group to SSH to VM
ingress_rule_2 = security_group.authorize(
self.apiclient,
- self.services["security_group_2"],
- account=self.account.account.name,
+ self.services["security_group_2"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.assertEqual(
@@ -1380,16 +1380,16 @@ class TestIngressRule(cloudstackTestCase):
self.services["security_group"]["endport"]
))
self.virtual_machine.get_ssh_client()
-
+
except Exception as e:
self.fail("SSH access failed for ingress rule ID: %s, %s" \
% (ingress_rule_1["id"], e))
-
+
# User should be able to ping VM
try:
self.debug("Trying to ping VM %s" % self.virtual_machine.ssh_ip)
result = subprocess.call(['ping', '-c 1', self.virtual_machine.ssh_ip])
-
+
self.debug("Ping result: %s" % result)
# if ping successful, then result should be 0
self.assertEqual(
@@ -1397,7 +1397,7 @@ class TestIngressRule(cloudstackTestCase):
0,
"Check if ping is successful or not"
)
-
+
except Exception as e:
self.fail("Ping failed for ingress rule ID: %s, %s" \
% (ingress_rule_2["id"], e))
@@ -1405,9 +1405,9 @@ class TestIngressRule(cloudstackTestCase):
def test_02_revokeIngressRule_AfterDeployVM(self):
"""Test Revoke ingress rule after deploy VM"""
-
+
tags = ["sg", "eip"]
-
+
# Validate the following
# 1. createsecuritygroup (ssh-incoming, 22via22) for this account
# 2. authorizeSecurityGroupIngress to allow ssh access to the VM
@@ -1420,13 +1420,13 @@ class TestIngressRule(cloudstackTestCase):
# but allowed through port 22
security_group = SecurityGroup.create(
- self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.apiclient,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.debug("Created security group with ID: %s" % security_group.id)
-
+
# Default Security group should not have any ingress rule
sercurity_groups = SecurityGroup.list(
self.apiclient,
@@ -1439,23 +1439,23 @@ class TestIngressRule(cloudstackTestCase):
"Check for list security groups response"
)
self.assertEqual(
- len(sercurity_groups),
- 2,
+ len(sercurity_groups),
+ 2,
"Check List Security groups response"
)
-
+
self.debug(
"Authorize Ingress Rule for Security Group %s for account: %s" \
% (
security_group.id,
self.account.account.name
))
-
+
# Authorize Security group to SSH to VM
ingress_rule = security_group.authorize(
self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.assertEqual(
@@ -1472,19 +1472,19 @@ class TestIngressRule(cloudstackTestCase):
securitygroupids=[security_group.id]
)
self.debug("Deploying VM in account: %s" % self.account.account.name)
-
+
self.debug(
"Authorize Ingress Rule for Security Group %s for account: %s" \
% (
security_group.id,
self.account.account.name
))
-
+
# Authorize Security group to SSH to VM
ingress_rule_2 = security_group.authorize(
self.apiclient,
- self.services["security_group_2"],
- account=self.account.account.name,
+ self.services["security_group_2"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.assertEqual(
@@ -1492,10 +1492,10 @@ class TestIngressRule(cloudstackTestCase):
True,
"Check ingress rule created properly"
)
-
+
ssh_rule = (ingress_rule["ingressrule"][0]).__dict__
icmp_rule = (ingress_rule_2["ingressrule"][0]).__dict__
-
+
# SSH should be allowed on 22
try:
self.debug("Trying to SSH into VM %s on port %s" % (
@@ -1503,16 +1503,16 @@ class TestIngressRule(cloudstackTestCase):
self.services["security_group"]["endport"]
))
self.virtual_machine.get_ssh_client()
-
+
except Exception as e:
self.fail("SSH access failed for ingress rule ID: %s, %s" \
% (ssh_rule["ruleid"], e))
-
+
# User should be able to ping VM
try:
self.debug("Trying to ping VM %s" % self.virtual_machine.ssh_ip)
result = subprocess.call(['ping', '-c 1', self.virtual_machine.ssh_ip])
-
+
self.debug("Ping result: %s" % result)
# if ping successful, then result should be 0
self.assertEqual(
@@ -1520,21 +1520,21 @@ class TestIngressRule(cloudstackTestCase):
0,
"Check if ping is successful or not"
)
-
+
except Exception as e:
self.fail("Ping failed for ingress rule ID: %s, %s" \
% (icmp_rule["ruleid"], e))
-
+
self.debug(
"Revoke Ingress Rule for Security Group %s for account: %s" \
% (
security_group.id,
self.account.account.name
))
-
+
result = security_group.revoke(
- self.apiclient,
- id = icmp_rule["ruleid"]
+ self.apiclient,
+ id=icmp_rule["ruleid"]
)
self.debug("Revoke ingress rule result: %s" % result)
@@ -1543,7 +1543,7 @@ class TestIngressRule(cloudstackTestCase):
try:
self.debug("Trying to ping VM %s" % self.virtual_machine.ssh_ip)
result = subprocess.call(['ping', '-c 1', self.virtual_machine.ssh_ip])
-
+
self.debug("Ping result: %s" % result)
# if ping successful, then result should be 0
self.assertNotEqual(
@@ -1551,7 +1551,7 @@ class TestIngressRule(cloudstackTestCase):
0,
"Check if ping is successful or not"
)
-
+
except Exception as e:
self.fail("Ping failed for ingress rule ID: %s, %s" \
% (icmp_rule["ruleid"], e))
@@ -1559,9 +1559,9 @@ class TestIngressRule(cloudstackTestCase):
def test_03_stopStartVM_verifyIngressAccess(self):
"""Test Start/Stop VM and Verify ingress rule"""
-
+
tags = ["sg", "eip"]
-
+
# Validate the following
# 1. createsecuritygroup (ssh-incoming, 22via22) for this account
# 2. authorizeSecurityGroupIngress to allow ssh access to the VM
@@ -1572,9 +1572,9 @@ class TestIngressRule(cloudstackTestCase):
# verify that ssh-access to the VM is allowed
security_group = SecurityGroup.create(
- self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.apiclient,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.debug("Created security group with ID: %s" % security_group.id)
@@ -1589,25 +1589,25 @@ class TestIngressRule(cloudstackTestCase):
True,
"Check for list security groups response"
)
-
+
self.assertEqual(
- len(sercurity_groups),
- 2,
+ len(sercurity_groups),
+ 2,
"Check List Security groups response"
)
-
+
self.debug(
"Authorize Ingress Rule for Security Group %s for account: %s" \
% (
security_group.id,
self.account.account.name
))
-
+
# Authorize Security group to SSH to VM
ingress_rule = security_group.authorize(
self.apiclient,
- self.services["security_group"],
- account=self.account.account.name,
+ self.services["security_group"],
+ account=self.account.account.name,
domainid=self.account.account.domainid
)
self.assertEqual(
@@ -1615,7 +1615,7 @@ class TestIngressRule(cloudstackTestCase):
True,
"Check ingress rule created properly"
)
-
+
self.virtual_machine = VirtualMachine.create(
self.apiclient,
self.services["virtual_machine"],
@@ -1625,7 +1625,7 @@ class TestIngressRule(cloudstackTestCase):
securitygroupids=[security_group.id]
)
self.debug("Deploying VM in account: %s" % self.account.account.name)
-
+
# SSH should be allowed on 22 port
try:
self.debug("Trying to SSH into VM %s" % self.virtual_machine.ssh_ip)
@@ -1634,17 +1634,17 @@ class TestIngressRule(cloudstackTestCase):
self.fail("SSH access failed for ingress rule ID: %s" \
% ingress_rule["id"]
)
-
+
self.virtual_machine.stop(self.apiclient)
-
+
# Sleep to ensure that VM is in stopped state
time.sleep(self.services["sleep"])
-
+
self.virtual_machine.start(self.apiclient)
-
+
# Sleep to ensure that VM is in running state
time.sleep(self.services["sleep"])
-
+
# SSH should be allowed on 22 port after restart
try:
self.debug("Trying to SSH into VM %s" % self.virtual_machine.ssh_ip)
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/b433b1af/test/integration/component/test_snapshots.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_snapshots.py b/test/integration/component/test_snapshots.py
index 9273919..dc98983 100644
--- a/test/integration/component/test_snapshots.py
+++ b/test/integration/component/test_snapshots.py
@@ -25,6 +25,7 @@ from integration.lib.base import *
from integration.lib.common import *
from marvin.remoteSSHClient import remoteSSHClient
+
class Services:
"""Test Snapshots Services
"""
@@ -44,8 +45,8 @@ class Services:
"name": "Tiny Instance",
"displaytext": "Tiny Instance",
"cpunumber": 1,
- "cpuspeed": 200, # in MHz
- "memory": 256, # In MBs
+ "cpuspeed": 200, # in MHz
+ "memory": 256, # In MBs
},
"disk_offering": {
"displaytext": "Small Disk",
@@ -71,10 +72,10 @@ class Services:
"recurring_snapshot": {
"intervaltype": 'HOURLY',
# Frequency of snapshots
- "maxsnaps": 1, # Should be min 2
+ "maxsnaps": 1, # Should be min 2
"schedule": 1,
"timezone": 'US/Arizona',
- # Timezone Formats - http://cloud.mindtouch.us/CloudStack_Documentation/Developer's_Guide%3A_CloudStack
+ # Timezone Formats - http://cloud.mindtouch.us/CloudStack_Documentation/Developer's_Guide%3A_CloudStack
},
"templates": {
"displaytext": 'Template',
@@ -84,7 +85,7 @@ class Services:
},
"diskdevice": "/dev/xvda",
"diskname": "TestDiskServ",
- "size": 1, # GBs
+ "size": 1, # GBs
"mount_dir": "/mnt/tmp",
"sub_dir": "test",
@@ -96,7 +97,7 @@ class Services:
# Cent OS 5.3 (64 bit)
"sleep": 60,
"timeout": 10,
- "mode" : 'advanced', # Networking mode: Advanced, Basic
+ "mode": 'advanced', # Networking mode: Advanced, Basic
}
@@ -165,7 +166,7 @@ class TestCreateVMsnapshotTemplate(cloudstackTestCase):
def test_01_createVM_snapshotTemplate(self):
"""Test create VM, Snapshot and Template
"""
- tags = ["advanced","advancedns"]
+ tags = ["advanced", "advancedns"]
# Validate the following
# 1. Deploy VM using default template, small service offering
# and small data disk offering.
@@ -191,7 +192,7 @@ class TestCreateVMsnapshotTemplate(cloudstackTestCase):
serviceofferingid=self.service_offering.id
)
self.debug("Created VM with ID: %s" % self.virtual_machine.id)
- # Get the Root disk of VM
+ # Get the Root disk of VM
volumes = list_volumes(
self.apiclient,
virtualmachineid=self.virtual_machine.id,
@@ -332,7 +333,7 @@ class TestCreateVMsnapshotTemplate(cloudstackTestCase):
export_path = '/'.join(parse_url[3:])
# Export path: export/test
-
+
# Sleep to ensure that snapshot is reflected in sec storage
time.sleep(self.services["sleep"])
try:
@@ -344,7 +345,7 @@ class TestCreateVMsnapshotTemplate(cloudstackTestCase):
self.services["mgmt_server"]["password"],
)
- cmds = [
+ cmds = [
"mkdir -p %s" % self.services["mount_dir"],
"mount %s/%s %s" % (
sec_storage_ip,
@@ -361,7 +362,7 @@ class TestCreateVMsnapshotTemplate(cloudstackTestCase):
self.debug("command: %s" % c)
result = ssh_client.execute(c)
self.debug("Result: %s" % result)
-
+
except Exception as e:
self.fail("SSH failed for Management server: %s" %
self.services["mgmt_server"]["ipaddress"])
@@ -379,7 +380,7 @@ class TestCreateVMsnapshotTemplate(cloudstackTestCase):
except Exception as e:
self.fail("SSH failed for Management server: %s" %
self.services["mgmt_server"]["ipaddress"])
-
+
res = str(uuids)
self.assertEqual(
res.count(snapshot_uuid),
@@ -430,7 +431,7 @@ class TestAccountSnapshotClean(cloudstackTestCase):
domainid=cls.account.account.domainid,
serviceofferingid=cls.service_offering.id
)
- # Get the Root disk of VM
+ # Get the Root disk of VM
volumes = list_volumes(
cls.api_client,
virtualmachineid=cls.virtual_machine.id,
@@ -473,7 +474,7 @@ class TestAccountSnapshotClean(cloudstackTestCase):
def test_02_accountSnapshotClean(self):
"""Test snapshot cleanup after account deletion
"""
- tags = ["advanced","advancedns"]
+ tags = ["advanced", "advancedns"]
# Validate the following
# 1. listAccounts API should list out the newly created account
# 2. listVirtualMachines() command should return the deployed VM.
@@ -518,7 +519,7 @@ class TestAccountSnapshotClean(cloudstackTestCase):
for virtual_machine in virtual_machines:
self.debug("VM ID: %s, VM state: %s" % (
virtual_machine.id,
- virtual_machine.state
+ virtual_machine.state
))
self.assertEqual(
virtual_machine.state,
@@ -591,7 +592,7 @@ class TestAccountSnapshotClean(cloudstackTestCase):
export_path = '/'.join(parse_url[3:])
# Export path: export/test
-
+
# Sleep to ensure that snapshot is reflected in sec storage
time.sleep(self.services["sleep"])
try:
@@ -621,7 +622,7 @@ class TestAccountSnapshotClean(cloudstackTestCase):
self.debug("command: %s" % c)
result = ssh_client.execute(c)
self.debug("Result: %s" % result)
-
+
uuids.append(result)
# Unmount the Sec Storage
@@ -633,14 +634,14 @@ class TestAccountSnapshotClean(cloudstackTestCase):
except Exception:
self.fail("SSH failed for management server: %s" %
self.services["mgmt_server"]["ipaddress"])
-
+
res = str(uuids)
self.assertEqual(
res.count(snapshot_uuid),
1,
"Check snapshot UUID in secondary storage and database"
)
-
+
self.debug("Deleting account: %s" % self.account.account.name)
# Delete account
self.account.delete(self.apiclient)
@@ -655,7 +656,7 @@ class TestAccountSnapshotClean(cloudstackTestCase):
"Check list response returns a valid list"
)
self.debug("account.cleanup.interval: %s" % interval[0].value)
-
+
# Wait for account cleanup interval
time.sleep(int(interval[0].value) * 2)
@@ -663,14 +664,14 @@ class TestAccountSnapshotClean(cloudstackTestCase):
self.apiclient,
id=self.account.account.id
)
-
- self.assertEqual(
- accounts,
- None,
- "List accounts should return empty list after account deletion"
- )
-
- uuids = []
+
+ self.assertEqual(
+ accounts,
+ None,
+ "List accounts should return empty list after account deletion"
+ )
+
+ uuids = []
for host in hosts:
# hosts[0].name = "nfs://192.168.100.21/export/test"
parse_url = (host.name).split('/')
@@ -684,7 +685,7 @@ class TestAccountSnapshotClean(cloudstackTestCase):
# Export path: export/test
try:
- cmds = [
+ cmds = [
"mount %s/%s %s" % (
sec_storage_ip,
export_path,
@@ -701,7 +702,7 @@ class TestAccountSnapshotClean(cloudstackTestCase):
self.debug("command: %s" % c)
result = ssh_client.execute(c)
self.debug("Result: %s" % result)
-
+
uuids.append(result)
# Unmount the Sec Storage
cmds = [
@@ -715,7 +716,7 @@ class TestAccountSnapshotClean(cloudstackTestCase):
except Exception:
self.fail("SSH failed for management server: %s" %
self.services["mgmt_server"]["ipaddress"])
-
+
res = str(uuids)
self.assertNotEqual(
res.count(snapshot_uuid),
@@ -804,7 +805,7 @@ class TestSnapshotDetachedDisk(cloudstackTestCase):
def test_03_snapshot_detachedDisk(self):
"""Test snapshot from detached disk
"""
- tags = ["advanced","advancedns"]
+ tags = ["advanced", "advancedns"]
# Validate the following
# 1. login in VM and write some data on data disk(use fdisk to
# partition datadisk,fdisk /dev/sdb, and make filesystem using
@@ -863,7 +864,7 @@ class TestSnapshotDetachedDisk(cloudstackTestCase):
self.services["sub_lvl_dir2"],
self.services["random_data"]
),
- "sync",
+ "sync",
]
for c in cmds:
self.debug(ssh_client.execute(c))
@@ -908,7 +909,7 @@ class TestSnapshotDetachedDisk(cloudstackTestCase):
except Exception as e:
self.fail("SSH failed for VM with IP: %s" %
self.virtual_machine.ipaddress)
-
+
# Fetch values from database
qresultset = self.dbclient.execute(
"select backup_snap_id, account_id, volume_id from snapshots where uuid = '%s';" \
@@ -954,7 +955,7 @@ class TestSnapshotDetachedDisk(cloudstackTestCase):
export_path = '/'.join(parse_url[3:])
# Export path: export/test
-
+
# Sleep to ensure that snapshot is reflected in sec storage
time.sleep(self.services["sleep"])
try:
@@ -983,7 +984,7 @@ class TestSnapshotDetachedDisk(cloudstackTestCase):
for c in cmds:
result = ssh_client.execute(c)
-
+
uuids.append(result)
# Unmount the Sec Storage
cmds = [
@@ -994,7 +995,7 @@ class TestSnapshotDetachedDisk(cloudstackTestCase):
except Exception as e:
self.fail("SSH failed for management server: %s" %
self.services["mgmt_server"]["ipaddress"])
-
+
res = str(uuids)
self.assertEqual(
res.count(snapshot_uuid),
@@ -1076,15 +1077,15 @@ class TestSnapshotLimit(cloudstackTestCase):
def test_04_snapshot_limit(self):
"""Test snapshot limit in snapshot policies
"""
- tags = ["advanced","advancedns"]
+ tags = ["advanced", "advancedns"]
# Validate the following
# 1. Perform hourly recurring snapshot on the root disk of VM and keep
# the maxsnapshots as 1
- # 2. listSnapshots should list the snapshot that was created
- # snapshot folder in secondary storage should contain only one
+ # 2. listSnapshots should list the snapshot that was created
+ # snapshot folder in secondary storage should contain only one
# snapshot image(/secondary/snapshots/$accountid/$volumeid/)
- # Get the Root disk of VM
+ # Get the Root disk of VM
volumes = list_volumes(
self.apiclient,
virtualmachineid=self.virtual_machine.id,
@@ -1116,7 +1117,7 @@ class TestSnapshotLimit(cloudstackTestCase):
True,
"Check list response returns a valid list"
)
-
+
self.assertNotEqual(
snapshot_policy,
None,
@@ -1148,7 +1149,7 @@ class TestSnapshotLimit(cloudstackTestCase):
snapshottype='RECURRING',
listall=True
)
-
+
self.assertEqual(
isinstance(snapshots, list),
True,
@@ -1159,7 +1160,7 @@ class TestSnapshotLimit(cloudstackTestCase):
self.services["recurring_snapshot"]["maxsnaps"],
"Check maximum number of recurring snapshots retained"
)
- snapshot = snapshots[0]
+ snapshot = snapshots[0]
# Sleep to ensure that snapshot is reflected in sec storage
time.sleep(self.services["sleep"])
@@ -1328,13 +1329,13 @@ class TestSnapshotEvents(cloudstackTestCase):
def test_05_snapshot_events(self):
"""Test snapshot events
"""
- tags = ["advanced","advancedns"]
+ tags = ["advanced", "advancedns"]
# Validate the following
# 1. Perform snapshot on the root disk of this VM and check the events/alerts.
# 2. delete the snapshots and check the events/alerts
# 3. listEvents() shows created/deleted snapshot events
- # Get the Root disk of VM
+ # Get the Root disk of VM
volumes = list_volumes(
self.apiclient,
virtualmachineid=self.virtual_machine.id,
http://git-wip-us.apache.org/repos/asf/incubator-cloudstack/blob/b433b1af/test/integration/component/test_templates.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_templates.py b/test/integration/component/test_templates.py
index 8016a43..4d74e0c 100644
--- a/test/integration/component/test_templates.py
+++ b/test/integration/component/test_templates.py
@@ -49,8 +49,8 @@ class Services:
"name": "Tiny Instance",
"displaytext": "Tiny Instance",
"cpunumber": 1,
- "cpuspeed": 100, # in MHz
- "memory": 64, # In MBs
+ "cpuspeed": 100, # in MHz
+ "memory": 64, # In MBs
},
"disk_offering": {
"displaytext": "Small",
@@ -71,15 +71,15 @@ class Services:
"diskname": "Test Volume",
},
"templates": {
- # Configs for different Template formats
+ # Configs for different Template formats
# For Eg. raw image, zip etc
- 0:{
+ 0: {
"displaytext": "Public Template",
"name": "Public template",
"ostypeid": '5776c0d2-f331-42db-ba3a-29f1f8319bc9',
"url": "http://download.cloud.com/releases/2.0.0/UbuntuServer-10-04-64bit.vhd.bz2",
"hypervisor": 'XenServer',
- "format" : 'VHD',
+ "format": 'VHD',
"isfeatured": True,
"ispublic": True,
"isextractable": True,
@@ -92,11 +92,11 @@ class Services:
"templatefilter": 'self',
},
"templatefilter": 'self',
- "destzoneid": 2, # For Copy template (Destination zone)
+ "destzoneid": 2, # For Copy template (Destination zone)
"ostypeid": '5776c0d2-f331-42db-ba3a-29f1f8319bc9',
"sleep": 60,
"timeout": 10,
- "mode": 'advanced', # Networking mode: Advanced, basic
+ "mode": 'advanced', # Networking mode: Advanced, basic
}
@@ -162,7 +162,7 @@ class TestCreateTemplate(cloudstackTestCase):
def test_01_create_template(self):
"""Test create public & private template
"""
- tags = ["advanced","advancedns"]
+ tags = ["advanced", "advancedns"]
# Validate the following:
# 1. Upload a templates in raw img format. Create a Vm instances from
# raw img template.
@@ -212,7 +212,7 @@ class TestCreateTemplate(cloudstackTestCase):
break
elif timeout == 0:
raise Exception("List template failed!")
-
+
time.sleep(5)
timeout = timeout - 1
#Verify template response to check whether template added successfully
@@ -221,13 +221,13 @@ class TestCreateTemplate(cloudstackTestCase):
True,
"Check for list template response return valid data"
)
-
+
self.assertNotEqual(
len(list_template_response),
0,
"Check template available in List Templates"
)
-
+
template_response = list_template_response[0]
self.assertEqual(
template_response.isready,
@@ -314,10 +314,10 @@ class TestTemplates(cloudstackTestCase):
#Stop virtual machine
cls.virtual_machine.stop(cls.api_client)
- timeout = cls.services["timeout"]
+ timeout = cls.services["timeout"]
#Wait before server has be successfully stopped
time.sleep(cls.services["sleep"])
-
+
while True:
list_volume = list_volumes(
cls.api_client,
@@ -329,10 +329,10 @@ class TestTemplates(cloudstackTestCase):
break
elif timeout == 0:
raise Exception("List volumes failed.")
-
+
time.sleep(5)
- timeout = timeout -1
-
+ timeout = timeout - 1
+
cls.volume = list_volume[0]
#Create template from volume
@@ -378,7 +378,7 @@ class TestTemplates(cloudstackTestCase):
def test_01_create_template_volume(self):
"""Test Create template from volume
"""
- tags = ["advanced","advancedns"]
+ tags = ["advanced", "advancedns"]
# Validate the following:
# 1. Deploy new VM using the template created from Volume
@@ -392,7 +392,7 @@ class TestTemplates(cloudstackTestCase):
domainid=self.account.account.domainid,
serviceofferingid=self.service_offering.id,
)
-
+
self.debug("creating an instance with template ID: %s" % self.template.id)
self.cleanup.append(virtual_machine)
vm_response = list_virtual_machines(
@@ -417,12 +417,12 @@ class TestTemplates(cloudstackTestCase):
def test_02_copy_template(self):
"""Test for copy template from one zone to another"""
- tags = ["advanced","advancedns"]
+ tags = ["advanced", "advancedns"]
# Validate the following
# 1. copy template should be successful and
# secondary storage should contain new copied template.
-
+
self.debug(
"Copying template from zone: %s to %s" % (
self.template.id,
@@ -447,7 +447,7 @@ class TestTemplates(cloudstackTestCase):
True,
"Check for list template response return valid list"
)
-
+
self.assertNotEqual(
len(list_template_response),
0,
@@ -476,7 +476,7 @@ class TestTemplates(cloudstackTestCase):
def test_03_delete_template(self):
"""Test Delete template
"""
- tags = ["advanced","advancedns"]
+ tags = ["advanced", "advancedns"]
# Validate the following:
# 1. Create a template and verify it is shown in list templates response
@@ -495,7 +495,7 @@ class TestTemplates(cloudstackTestCase):
True,
"Check for list template response return valid list"
)
-
+
self.assertNotEqual(
len(list_template_response),
0,
@@ -508,12 +508,12 @@ class TestTemplates(cloudstackTestCase):
self.template.id,
"Check display text of updated template"
)
-
+
self.debug("Deleting template: %s" % self.template)
# Delete the template
self.template.delete(self.apiclient)
self.debug("Delete template: %s successful" % self.template)
-
+
list_template_response = list_templates(
self.apiclient,
templatefilter=\
@@ -531,7 +531,7 @@ class TestTemplates(cloudstackTestCase):
def test_04_template_from_snapshot(self):
"""Create Template from snapshot
"""
- tags = ["advanced","advancedns"]
+ tags = ["advanced", "advancedns"]
# Validate the following
# 2. Snapshot the Root disk
@@ -546,7 +546,7 @@ class TestTemplates(cloudstackTestCase):
listall=True
)
volume = volumes[0]
-
+
self.debug("Creating a snapshot from volume: %s" % volume.id)
#Create a snapshot of volume
snapshot = Snapshot.create(