You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by ga...@apache.org on 2015/05/06 13:01:12 UTC
[1/4] git commit: updated refs/heads/master to 6c3dc15
Repository: cloudstack
Updated Branches:
refs/heads/master 464af9cb2 -> 6c3dc15fc
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/6c3dc15f/test/integration/component/test_vpc_routers.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_vpc_routers.py b/test/integration/component/test_vpc_routers.py
index 54f6a83..c27e4b2 100644
--- a/test/integration/component/test_vpc_routers.py
+++ b/test/integration/component/test_vpc_routers.py
@@ -15,155 +15,177 @@
# specific language governing permissions and limitations
# under the License.
-""" Component tests for VPC - Router Operations
+""" Component tests for VPC - Router Operations
"""
-#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 *
-import datetime
+from marvin.cloudstackTestCase import cloudstackTestCase
+from marvin.cloudstackAPI import (startRouter,
+ stopRouter,
+ rebootRouter,
+ migrateSystemVm)
+from marvin.lib.utils import cleanup_resources
+from marvin.lib.base import (Account,
+ Router,
+ ServiceOffering,
+ VirtualMachine,
+ Host,
+ PublicIPAddress,
+ VPC,
+ VpcOffering,
+ StaticRoute,
+ Network,
+ NetworkOffering,
+ PrivateGateway,
+ NATRule,
+ NetworkACL,
+ StaticNATRule,
+ LoadBalancerRule)
+from marvin.lib.common import (get_domain,
+ get_zone,
+ list_routers,
+ get_template)
+import time
class Services:
+
"""Test VPC Router services
"""
def __init__(self):
self.services = {
- "account": {
- "email": "test@test.com",
- "firstname": "Test",
- "lastname": "User",
- "username": "test",
- # Random characters are appended for unique
- # username
- "password": "password",
- },
- "service_offering": {
- "name": "Tiny Instance",
- "displaytext": "Tiny Instance",
- "cpunumber": 1,
- "cpuspeed": 100,
- "memory": 128,
- },
- "service_offering_new": {
- "name": "Small Instance",
- "displaytext": "Small Instance",
- "cpunumber": 1,
- "cpuspeed": 100,
- "memory": 256,
- "issystem": 'true',
- },
-
- "network_offering": {
- "name": 'VPC Network offering',
- "displaytext": 'VPC Network off',
- "guestiptype": 'Isolated',
- "supportedservices": 'Vpn,Dhcp,Dns,SourceNat,PortForwarding,Lb,UserData,StaticNat,NetworkACL',
- "traffictype": 'GUEST',
- "availability": 'Optional',
- "useVpc": 'on',
- "serviceProviderList": {
- "Vpn": 'VpcVirtualRouter',
- "Dhcp": 'VpcVirtualRouter',
- "Dns": 'VpcVirtualRouter',
- "SourceNat": 'VpcVirtualRouter',
- "PortForwarding": 'VpcVirtualRouter',
- "Lb": 'VpcVirtualRouter',
- "UserData": 'VpcVirtualRouter',
- "StaticNat": 'VpcVirtualRouter',
- "NetworkACL": 'VpcVirtualRouter'
- },
- },
- "network_offering_no_lb": {
- "name": 'VPC Network offering',
- "displaytext": 'VPC Network off',
- "guestiptype": 'Isolated',
- "supportedservices": 'Vpn,Dhcp,Dns,SourceNat,PortForwarding,UserData,StaticNat,NetworkACL',
- "traffictype": 'GUEST',
- "availability": 'Optional',
- "useVpc": 'on',
- "serviceProviderList": {
- "Vpn": 'VpcVirtualRouter',
- "Dhcp": 'VpcVirtualRouter',
- "Dns": 'VpcVirtualRouter',
- "SourceNat": 'VpcVirtualRouter',
- "PortForwarding": 'VpcVirtualRouter',
- "UserData": 'VpcVirtualRouter',
- "StaticNat": 'VpcVirtualRouter',
- "NetworkACL": 'VpcVirtualRouter'
- },
- },
- "vpc_offering": {
- "name": 'VPC off',
- "displaytext": 'VPC off',
- "supportedservices": 'Dhcp,Dns,SourceNat,PortForwarding,Vpn,Lb,UserData,StaticNat',
- },
- "vpc": {
- "name": "TestVPC",
- "displaytext": "TestVPC",
- "cidr": '10.0.0.1/24'
- },
- "network": {
- "name": "Test Network",
- "displaytext": "Test Network",
- "netmask": '255.255.255.0'
- },
- "lbrule": {
- "name": "SSH",
- "alg": "leastconn",
- # Algorithm used for load balancing
- "privateport": 22,
- "publicport": 2222,
- "openfirewall": False,
- "startport": 2222,
- "endport": 2222,
- "protocol": "TCP",
- "cidrlist": '0.0.0.0/0',
- },
- "natrule": {
- "privateport": 22,
- "publicport": 22,
- "startport": 22,
- "endport": 22,
- "protocol": "TCP",
- "cidrlist": '0.0.0.0/0',
- },
- "fw_rule": {
- "startport": 1,
- "endport": 6000,
- "cidr": '0.0.0.0/0',
- # Any network (For creating FW rule)
- "protocol": "TCP"
- },
- "http_rule": {
- "startport": 80,
- "endport": 80,
- "cidrlist": '0.0.0.0/0',
- "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,
- "mode": 'advanced'
- }
+ "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,
+ "memory": 128,
+ },
+ "service_offering_new": {
+ "name": "Small Instance",
+ "displaytext": "Small Instance",
+ "cpunumber": 1,
+ "cpuspeed": 100,
+ "memory": 256,
+ "issystem": 'true',
+ },
+
+ "network_offering": {
+ "name": 'VPC Network offering',
+ "displaytext": 'VPC Network off',
+ "guestiptype": 'Isolated',
+ "supportedservices": 'Vpn,Dhcp,Dns,SourceNat,PortForwarding,Lb,UserData,StaticNat,NetworkACL',
+ "traffictype": 'GUEST',
+ "availability": 'Optional',
+ "useVpc": 'on',
+ "serviceProviderList": {
+ "Vpn": 'VpcVirtualRouter',
+ "Dhcp": 'VpcVirtualRouter',
+ "Dns": 'VpcVirtualRouter',
+ "SourceNat": 'VpcVirtualRouter',
+ "PortForwarding": 'VpcVirtualRouter',
+ "Lb": 'VpcVirtualRouter',
+ "UserData": 'VpcVirtualRouter',
+ "StaticNat": 'VpcVirtualRouter',
+ "NetworkACL": 'VpcVirtualRouter'
+ },
+ },
+ "network_offering_no_lb": {
+ "name": 'VPC Network offering',
+ "displaytext": 'VPC Network off',
+ "guestiptype": 'Isolated',
+ "supportedservices": 'Vpn,Dhcp,Dns,SourceNat,PortForwarding,UserData,StaticNat,NetworkACL',
+ "traffictype": 'GUEST',
+ "availability": 'Optional',
+ "useVpc": 'on',
+ "serviceProviderList": {
+ "Vpn": 'VpcVirtualRouter',
+ "Dhcp": 'VpcVirtualRouter',
+ "Dns": 'VpcVirtualRouter',
+ "SourceNat": 'VpcVirtualRouter',
+ "PortForwarding": 'VpcVirtualRouter',
+ "UserData": 'VpcVirtualRouter',
+ "StaticNat": 'VpcVirtualRouter',
+ "NetworkACL": 'VpcVirtualRouter'
+ },
+ },
+ "vpc_offering": {
+ "name": 'VPC off',
+ "displaytext": 'VPC off',
+ "supportedservices": 'Dhcp,Dns,SourceNat,PortForwarding,Vpn,Lb,UserData,StaticNat',
+ },
+ "vpc": {
+ "name": "TestVPC",
+ "displaytext": "TestVPC",
+ "cidr": '10.0.0.1/24'
+ },
+ "network": {
+ "name": "Test Network",
+ "displaytext": "Test Network",
+ "netmask": '255.255.255.0'
+ },
+ "lbrule": {
+ "name": "SSH",
+ "alg": "leastconn",
+ # Algorithm used for load balancing
+ "privateport": 22,
+ "publicport": 2222,
+ "openfirewall": False,
+ "startport": 2222,
+ "endport": 2222,
+ "protocol": "TCP",
+ "cidrlist": '0.0.0.0/0',
+ },
+ "natrule": {
+ "privateport": 22,
+ "publicport": 22,
+ "startport": 22,
+ "endport": 22,
+ "protocol": "TCP",
+ "cidrlist": '0.0.0.0/0',
+ },
+ "fw_rule": {
+ "startport": 1,
+ "endport": 6000,
+ "cidr": '0.0.0.0/0',
+ # Any network (For creating FW rule)
+ "protocol": "TCP"
+ },
+ "http_rule": {
+ "startport": 80,
+ "endport": 80,
+ "cidrlist": '0.0.0.0/0',
+ "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,
+ "mode": 'advanced'
+ }
+
class TestVPCRoutersBasic(cloudstackTestCase):
@@ -177,44 +199,44 @@ class TestVPCRoutersBasic(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.template = get_template(
- cls.api_client,
- cls.zone.id,
- cls.services["ostype"]
- )
+ cls.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.vpc_off = VpcOffering.create(
- cls.api_client,
- cls.services["vpc_offering"]
- )
+ cls.api_client,
+ cls.services["vpc_offering"]
+ )
cls.vpc_off.update(cls.api_client, state='Enabled')
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
+ )
cls._cleanup = [cls.account]
cls._cleanup.append(cls.vpc_off)
#cls.debug("Enabling the VPC offering created")
cls.vpc_off.update(cls.api_client, state='Enabled')
- #cls.debug("creating a VPC network in the account: %s" %
- # cls.account.name)
+ # cls.debug("creating a VPC network in the account: %s" %
+ # cls.account.name)
cls.services["vpc"]["cidr"] = '10.1.1.1/16'
cls.vpc = VPC.create(
- cls.api_client,
- cls.services["vpc"],
- vpcofferingid=cls.vpc_off.id,
- zoneid=cls.zone.id,
- account=cls.account.name,
- domainid=cls.account.domainid
- )
+ cls.api_client,
+ cls.services["vpc"],
+ vpcofferingid=cls.vpc_off.id,
+ zoneid=cls.zone.id,
+ account=cls.account.name,
+ domainid=cls.account.domainid
+ )
cls._cleanup.append(cls.service_offering)
return
@@ -222,7 +244,7 @@ class TestVPCRoutersBasic(cloudstackTestCase):
@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)
@@ -230,33 +252,33 @@ class TestVPCRoutersBasic(cloudstackTestCase):
def setUp(self):
self.api_client = self.testClient.getApiClient()
-
+
return
def tearDown(self):
- return
+ return
def validate_vpc_offering(self, vpc_offering):
"""Validates the VPC offering"""
self.debug("Check if the VPC offering is created successfully?")
vpc_offs = VpcOffering.list(
- self.api_client,
- id=vpc_offering.id
- )
+ self.api_client,
+ id=vpc_offering.id
+ )
self.assertEqual(
- isinstance(vpc_offs, list),
- True,
- "List VPC offerings should return a valid list"
- )
+ isinstance(vpc_offs, list),
+ True,
+ "List VPC offerings should return a valid list"
+ )
self.assertEqual(
- vpc_offering.name,
- vpc_offs[0].name,
- "Name of the VPC offering should match with listVPCOff data"
- )
+ vpc_offering.name,
+ vpc_offs[0].name,
+ "Name of the VPC offering should match with listVPCOff data"
+ )
self.debug(
- "VPC offering is created successfully - %s" %
- vpc_offering.name)
+ "VPC offering is created successfully - %s" %
+ vpc_offering.name)
return
def validate_vpc_network(self, network, state=None):
@@ -264,28 +286,28 @@ class TestVPCRoutersBasic(cloudstackTestCase):
self.debug("Check if the VPC network is created successfully?")
vpc_networks = VPC.list(
- self.api_client,
- id=network.id
- )
+ self.api_client,
+ id=network.id
+ )
self.assertEqual(
- isinstance(vpc_networks, list),
- True,
- "List VPC network should return a valid list"
- )
+ isinstance(vpc_networks, list),
+ True,
+ "List VPC network should return a valid list"
+ )
self.assertEqual(
- network.name,
- vpc_networks[0].name,
- "Name of the VPC network should match with listVPC data"
- )
+ network.name,
+ vpc_networks[0].name,
+ "Name of the VPC network should match with listVPC data"
+ )
if state:
self.assertEqual(
- vpc_networks[0].state,
- state,
+ vpc_networks[0].state,
+ state,
"VPC state should be '%s'" % state
- )
+ )
self.debug("VPC network validated - %s" % network.name)
return
-
+
def migrate_router(self, router):
""" Migrate the router """
@@ -293,13 +315,13 @@ class TestVPCRoutersBasic(cloudstackTestCase):
hosts = Host.list(self.api_client, zoneid=self.zone.id, type='Routing')
self.assertEqual(
- isinstance(hosts, list),
- True,
- "List hosts should return a valid list"
- )
+ isinstance(hosts, list),
+ True,
+ "List hosts should return a valid list"
+ )
if len(hosts) < 2:
- raise unittest.SkipTest(
- "No host available for migration. Test requires atleast 2 hosts")
+ self.skipTest(
+ "No host available for migration. Test requires atleast 2 hosts")
# Remove the host of current VM from the hosts list
hosts[:] = [host for host in hosts if host.id != router.hostid]
@@ -307,13 +329,13 @@ class TestVPCRoutersBasic(cloudstackTestCase):
self.debug("Validating if the network rules work properly or not?")
self.debug("Migrating VM-ID: %s from %s to Host: %s" % (
- router.id,
- router.hostid,
- host.id
- ))
+ router.id,
+ router.hostid,
+ host.id
+ ))
try:
- #Migrate the router
+ # Migrate the router
cmd = migrateSystemVm.migrateSystemVmCmd()
cmd.isAsync = "false"
cmd.hostid = host.id
@@ -322,102 +344,104 @@ class TestVPCRoutersBasic(cloudstackTestCase):
except Exception as e:
self.fail("Failed to migrate instance, %s" % e)
-
+
self.debug("Waiting for Router mgiration ....")
- time.sleep(240)
-
- #List routers to check state of router
+ time.sleep(240)
+
+ # List routers to check state of router
router_response = list_routers(
- self.api_client,
- id=router.id
- )
+ self.api_client,
+ id=router.id
+ )
self.assertEqual(
- isinstance(router_response, list),
- True,
- "Check list response returns a valid list"
- )
-
+ isinstance(router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
+
router.hostid = router_response[0].hostid
- self.assertEqual(router.hostid, host.id, "Migration to host %s failed. The router host is"
- " still %s" % (host.id, router.hostid))
+ self.assertEqual(
+ router.hostid, host.id, "Migration to host %s failed. The router host is"
+ " still %s" %
+ (host.id, router.hostid))
return
@attr(tags=["advanced", "intervlan"], required_hardware="false")
def test_01_stop_start_router_after_creating_vpc(self):
""" Test to stop and start router after creation of VPC
"""
-
- # Validate following:
- # 1. Create a VPC with cidr - 10.1.1.1/16
- # 2. Stop the VPC Virtual Router which is created as a result of VPC creation.
- # 3. Start the Stopped VPC Virtual Router
+
+ # Validate following:
+ # 1. Create a VPC with cidr - 10.1.1.1/16
+ # 2. Stop the VPC Virtual Router which is created as a result of VPC creation.
+ # 3. Start the Stopped VPC Virtual Router
self.validate_vpc_offering(self.vpc_off)
self.validate_vpc_network(self.vpc)
- # Stop the VPC Router
+ # Stop the VPC Router
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
- True,
- "List Routers should return a valid list"
- )
- router = routers[0]
+ isinstance(routers, list),
+ True,
+ "List Routers should return a valid list"
+ )
+ router = routers[0]
self.debug("Stopping the router with ID: %s" % router.id)
- #Stop the router
+ # Stop the router
cmd = stopRouter.stopRouterCmd()
cmd.id = router.id
self.api_client.stopRouter(cmd)
-
- #List routers to check state of router
+
+ # List routers to check state of router
router_response = list_routers(
- self.api_client,
- id=router.id
- )
+ self.api_client,
+ id=router.id
+ )
self.assertEqual(
- isinstance(router_response, list),
- True,
- "Check list response returns a valid list"
- )
- #List router should have router in stopped state
+ isinstance(router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
+ # List router should have router in stopped state
self.assertEqual(
- router_response[0].state,
- 'Stopped',
- "Check list router response for router state"
- )
+ router_response[0].state,
+ 'Stopped',
+ "Check list router response for router state"
+ )
self.debug("Stopped the router with ID: %s" % router.id)
- # Start The Router
+ # Start The Router
self.debug("Starting the router with ID: %s" % router.id)
cmd = startRouter.startRouterCmd()
cmd.id = router.id
self.api_client.startRouter(cmd)
- #List routers to check state of router
+ # List routers to check state of router
router_response = list_routers(
- self.api_client,
- id=router.id
- )
+ self.api_client,
+ id=router.id
+ )
self.assertEqual(
- isinstance(router_response, list),
- True,
- "Check list response returns a valid list"
- )
- #List router should have router in running state
+ isinstance(router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
+ # List router should have router in running state
self.assertEqual(
- router_response[0].state,
- 'Running',
- "Check list router response for router state"
- )
+ router_response[0].state,
+ 'Running',
+ "Check list router response for router state"
+ )
self.debug("Started the router with ID: %s" % router.id)
-
+
return
@attr(tags=["advanced", "intervlan"], required_hardware="false")
@@ -425,71 +449,72 @@ class TestVPCRoutersBasic(cloudstackTestCase):
""" Test to reboot the router after creating a VPC
"""
# Validate the following
- # 1. Create a VPC with cidr - 10.1.1.1/16
- # 2. Reboot the VPC Virtual Router which is created as a result of VPC creation.
- # Stop the VPC Router
-
+ # 1. Create a VPC with cidr - 10.1.1.1/16
+ # 2. Reboot the VPC Virtual Router which is created as a result of VPC creation.
+ # Stop the VPC Router
+
self.validate_vpc_offering(self.vpc_off)
self.validate_vpc_network(self.vpc)
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
- True,
- "List Routers should return a valid list"
- )
- router = routers[0]
+ isinstance(routers, list),
+ True,
+ "List Routers should return a valid list"
+ )
+ router = routers[0]
self.debug("Rebooting the router ...")
- #Reboot the router
+ # Reboot the router
cmd = rebootRouter.rebootRouterCmd()
cmd.id = router.id
self.api_client.rebootRouter(cmd)
- #List routers to check state of router
+ # List routers to check state of router
router_response = list_routers(
- self.api_client,
- id=router.id
- )
+ self.api_client,
+ id=router.id
+ )
self.assertEqual(
- isinstance(router_response, list),
- True,
- "Check list response returns a valid list"
- )
- #List router should have router in running state and same public IP
+ isinstance(router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
+ # List router should have router in running state and same public IP
self.assertEqual(
- router_response[0].state,
- 'Running',
- "Check list router response for router state"
- )
+ router_response[0].state,
+ 'Running',
+ "Check list router response for router state"
+ )
return
-
@attr(tags=["advanced", "intervlan"], required_hardware="true")
def test_03_migrate_router_after_creating_vpc(self):
""" Test migration of router to another host after creating VPC """
self.hypervisor = self.testClient.getHypervisorInfo()
if self.hypervisor.lower() in ['lxc']:
- self.skipTest("vm migrate is not supported in %s" % self.hypervisor)
+ self.skipTest(
+ "vm migrate is not supported in %s" %
+ self.hypervisor)
self.validate_vpc_offering(self.vpc_off)
self.validate_vpc_network(self.vpc)
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
- True,
- "List Routers should return a valid list"
- )
+ isinstance(routers, list),
+ True,
+ "List Routers should return a valid list"
+ )
self.migrate_router(routers[0])
return
@@ -498,27 +523,28 @@ class TestVPCRoutersBasic(cloudstackTestCase):
""" Tests to change service offering of the Router after
creating a vpc
"""
-
+
# Validate the following
- # 1. Create a VPC with cidr - 10.1.1.1/16
- # 2. Change the service offerings of the VPC Virtual Router which is created as a result of VPC creation.
-
+ # 1. Create a VPC with cidr - 10.1.1.1/16
+ # 2. Change the service offerings of the VPC Virtual Router which is
+ # created as a result of VPC creation.
+
self.validate_vpc_offering(self.vpc_off)
self.validate_vpc_network(self.vpc)
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
- True,
- "List Routers should return a valid list"
- )
-
- #Stop the router
+ isinstance(routers, list),
+ True,
+ "List Routers should return a valid list"
+ )
+
+ # Stop the router
router = routers[0]
self.debug("Stopping the router with ID: %s" % router.id)
cmd = stopRouter.stopRouterCmd()
@@ -526,72 +552,74 @@ class TestVPCRoutersBasic(cloudstackTestCase):
self.api_client.stopRouter(cmd)
service_offering = ServiceOffering.create(
- self.api_client,
- self.services["service_offering_new"]
- )
+ self.api_client,
+ self.services["service_offering_new"]
+ )
self.debug("Changing service offering for the Router %s" % router.id)
- try:
+ try:
router = Router.change_service_offering(self.api_client,
- router.id,
- service_offering.id
- )
+ router.id,
+ service_offering.id
+ )
except:
self.fail("Changing service offering failed")
-
- self.debug("Router %s" % router)
+
+ self.debug("Router %s" % router)
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
router = routers[0]
self.assertEqual(
- router.serviceofferingid,
- service_offering.id,
- "Changing service offering failed as id is %s and expected"
- "is %s" % (router.serviceofferingid, service_offering.id)
- )
+ router.serviceofferingid,
+ service_offering.id,
+ "Changing service offering failed as id is %s and expected"
+ "is %s" % (router.serviceofferingid, service_offering.id)
+ )
return
@attr(tags=["advanced", "intervlan"], required_hardware="false")
def test_05_destroy_router_after_creating_vpc(self):
""" Test to destroy the router after creating a VPC
- """
+ """
# Validate the following
- # 1. Create a VPC with cidr - 10.1.1.1/16
- # 2. Destroy the VPC Virtual Router which is created as a result of VPC creation.
+ # 1. Create a VPC with cidr - 10.1.1.1/16
+ # 2. Destroy the VPC Virtual Router which is created as a result of VPC
+ # creation.
self.validate_vpc_offering(self.vpc_off)
self.validate_vpc_network(self.vpc)
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
- True,
- "List Routers should return a valid list"
- )
-
- Router.destroy( self.api_client,
- id=routers[0].id
- )
-
+ isinstance(routers, list),
+ True,
+ "List Routers should return a valid list"
+ )
+
+ Router.destroy(self.api_client,
+ id=routers[0].id
+ )
+
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
- False,
- "List Routers should be empty"
- )
+ isinstance(routers, list),
+ False,
+ "List Routers should be empty"
+ )
return
+
class TestVPCRouterOneNetwork(cloudstackTestCase):
@classmethod
@@ -605,216 +633,216 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.template = get_template(
- cls.api_client,
- cls.zone.id,
- cls.services["ostype"]
- )
+ cls.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._cleanup.append(cls.service_offering)
cls.vpc_off = VpcOffering.create(
- cls.api_client,
- cls.services["vpc_offering"]
- )
+ cls.api_client,
+ cls.services["vpc_offering"]
+ )
cls.vpc_off.update(cls.api_client, state='Enabled')
cls._cleanup.append(cls.vpc_off)
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
+ )
cls._cleanup.insert(0, cls.account)
cls.services["vpc"]["cidr"] = '10.1.1.1/16'
cls.vpc = VPC.create(
- cls.api_client,
- cls.services["vpc"],
- vpcofferingid=cls.vpc_off.id,
- zoneid=cls.zone.id,
- account=cls.account.name,
- domainid=cls.account.domainid
- )
+ cls.api_client,
+ cls.services["vpc"],
+ vpcofferingid=cls.vpc_off.id,
+ zoneid=cls.zone.id,
+ account=cls.account.name,
+ domainid=cls.account.domainid
+ )
private_gateway = PrivateGateway.create(
- cls.api_client,
- gateway='10.1.3.1',
- ipaddress='10.1.3.100',
- netmask='255.255.255.0',
- vlan=678,
- vpcid=cls.vpc.id
- )
+ cls.api_client,
+ gateway='10.1.3.1',
+ ipaddress='10.1.3.100',
+ netmask='255.255.255.0',
+ vlan=678,
+ vpcid=cls.vpc.id
+ )
cls.gateways = PrivateGateway.list(
- cls.api_client,
- id=private_gateway.id,
- listall=True
- )
+ cls.api_client,
+ id=private_gateway.id,
+ listall=True
+ )
static_route = StaticRoute.create(
- cls.api_client,
- cidr='11.1.1.1/24',
- gatewayid=private_gateway.id
- )
+ cls.api_client,
+ cidr='11.1.1.1/24',
+ gatewayid=private_gateway.id
+ )
cls.static_routes = StaticRoute.list(
- cls.api_client,
- id=static_route.id,
- listall=True
- )
-
+ cls.api_client,
+ id=static_route.id,
+ listall=True
+ )
+
cls.nw_off = NetworkOffering.create(
- cls.api_client,
- cls.services["network_offering"],
- conservemode=False
- )
+ cls.api_client,
+ cls.services["network_offering"],
+ conservemode=False
+ )
# Enable Network offering
cls.nw_off.update(cls.api_client, state='Enabled')
cls._cleanup.append(cls.nw_off)
# Creating network using the network offering created
cls.network_1 = Network.create(
- cls.api_client,
- cls.services["network"],
- accountid=cls.account.name,
- domainid=cls.account.domainid,
- networkofferingid=cls.nw_off.id,
- zoneid=cls.zone.id,
- gateway='10.1.1.1',
- vpcid=cls.vpc.id
- )
+ cls.api_client,
+ cls.services["network"],
+ accountid=cls.account.name,
+ domainid=cls.account.domainid,
+ networkofferingid=cls.nw_off.id,
+ zoneid=cls.zone.id,
+ gateway='10.1.1.1',
+ vpcid=cls.vpc.id
+ )
# Spawn an instance in that network
vm_1 = VirtualMachine.create(
- cls.api_client,
- cls.services["virtual_machine"],
- accountid=cls.account.name,
- domainid=cls.account.domainid,
- serviceofferingid=cls.service_offering.id,
- networkids=[str(cls.network_1.id)]
- )
+ cls.api_client,
+ cls.services["virtual_machine"],
+ accountid=cls.account.name,
+ domainid=cls.account.domainid,
+ serviceofferingid=cls.service_offering.id,
+ networkids=[str(cls.network_1.id)]
+ )
vm_2 = VirtualMachine.create(
- cls.api_client,
- cls.services["virtual_machine"],
- accountid=cls.account.name,
- domainid=cls.account.domainid,
- serviceofferingid=cls.service_offering.id,
- networkids=[str(cls.network_1.id)]
- )
+ cls.api_client,
+ cls.services["virtual_machine"],
+ accountid=cls.account.name,
+ domainid=cls.account.domainid,
+ serviceofferingid=cls.service_offering.id,
+ networkids=[str(cls.network_1.id)]
+ )
# Spawn an instance in that network
vm_3 = VirtualMachine.create(
- cls.api_client,
- cls.services["virtual_machine"],
- accountid=cls.account.name,
- domainid=cls.account.domainid,
- serviceofferingid=cls.service_offering.id,
- networkids=[str(cls.network_1.id)]
- )
-
- vms = VirtualMachine.list(
- cls.api_client,
- account=cls.account.name,
- domainid=cls.account.domainid,
- listall=True
- )
+ cls.api_client,
+ cls.services["virtual_machine"],
+ accountid=cls.account.name,
+ domainid=cls.account.domainid,
+ serviceofferingid=cls.service_offering.id,
+ networkids=[str(cls.network_1.id)]
+ )
+
+ VirtualMachine.list(
+ cls.api_client,
+ account=cls.account.name,
+ domainid=cls.account.domainid,
+ listall=True
+ )
public_ip_1 = PublicIPAddress.create(
- cls.api_client,
- accountid=cls.account.name,
- zoneid=cls.zone.id,
- domainid=cls.account.domainid,
- networkid=cls.network_1.id,
- vpcid=cls.vpc.id
- )
-
- nat_rule = NATRule.create(
- cls.api_client,
- vm_1,
- cls.services["natrule"],
- ipaddressid=public_ip_1.ipaddress.id,
- openfirewall=False,
- networkid=cls.network_1.id,
- vpcid=cls.vpc.id
- )
-
- nwacl_nat = NetworkACL.create(
- cls.api_client,
- networkid=cls.network_1.id,
- services=cls.services["natrule"],
- traffictype='Ingress'
- )
+ cls.api_client,
+ accountid=cls.account.name,
+ zoneid=cls.zone.id,
+ domainid=cls.account.domainid,
+ networkid=cls.network_1.id,
+ vpcid=cls.vpc.id
+ )
+
+ NATRule.create(
+ cls.api_client,
+ vm_1,
+ cls.services["natrule"],
+ ipaddressid=public_ip_1.ipaddress.id,
+ openfirewall=False,
+ networkid=cls.network_1.id,
+ vpcid=cls.vpc.id
+ )
+
+ NetworkACL.create(
+ cls.api_client,
+ networkid=cls.network_1.id,
+ services=cls.services["natrule"],
+ traffictype='Ingress'
+ )
public_ip_2 = PublicIPAddress.create(
- cls.api_client,
- accountid=cls.account.name,
- zoneid=cls.zone.id,
- domainid=cls.account.domainid,
- networkid=cls.network_1.id,
- vpcid=cls.vpc.id
- )
+ cls.api_client,
+ accountid=cls.account.name,
+ zoneid=cls.zone.id,
+ domainid=cls.account.domainid,
+ networkid=cls.network_1.id,
+ vpcid=cls.vpc.id
+ )
try:
StaticNATRule.enable(
- cls.api_client,
- ipaddressid=public_ip_2.ipaddress.id,
- virtualmachineid=vm_2.id,
- networkid=cls.network_1.id
- )
+ cls.api_client,
+ ipaddressid=public_ip_2.ipaddress.id,
+ virtualmachineid=vm_2.id,
+ networkid=cls.network_1.id
+ )
except Exception as e:
cls.fail("Failed to enable static NAT on IP: %s - %s" % (
- public_ip_2.ipaddress.ipaddress, e))
-
- public_ips = PublicIPAddress.list(
- cls.api_client,
- networkid=cls.network_1.id,
- listall=True,
- isstaticnat=True,
- account=cls.account.name,
- domainid=cls.account.domainid
- )
+ public_ip_2.ipaddress.ipaddress, e))
+
+ PublicIPAddress.list(
+ cls.api_client,
+ networkid=cls.network_1.id,
+ listall=True,
+ isstaticnat=True,
+ account=cls.account.name,
+ domainid=cls.account.domainid
+ )
public_ip_3 = PublicIPAddress.create(
- cls.api_client,
- accountid=cls.account.name,
- zoneid=cls.zone.id,
- domainid=cls.account.domainid,
- networkid=cls.network_1.id,
- vpcid=cls.vpc.id
- )
+ cls.api_client,
+ accountid=cls.account.name,
+ zoneid=cls.zone.id,
+ domainid=cls.account.domainid,
+ networkid=cls.network_1.id,
+ vpcid=cls.vpc.id
+ )
lb_rule = LoadBalancerRule.create(
- cls.api_client,
- cls.services["lbrule"],
- ipaddressid=public_ip_3.ipaddress.id,
- accountid=cls.account.name,
- networkid=cls.network_1.id,
- vpcid=cls.vpc.id,
- domainid=cls.account.domainid
- )
+ cls.api_client,
+ cls.services["lbrule"],
+ ipaddressid=public_ip_3.ipaddress.id,
+ accountid=cls.account.name,
+ networkid=cls.network_1.id,
+ vpcid=cls.vpc.id,
+ domainid=cls.account.domainid
+ )
lb_rule.assign(cls.api_client, [vm_3])
- nwacl_lb = NetworkACL.create(
- cls.api_client,
- networkid=cls.network_1.id,
- services=cls.services["lbrule"],
- traffictype='Ingress'
- )
-
- nwacl_internet_1 = NetworkACL.create(
- cls.api_client,
- networkid=cls.network_1.id,
- services=cls.services["http_rule"],
- traffictype='Egress'
- )
-
+ NetworkACL.create(
+ cls.api_client,
+ networkid=cls.network_1.id,
+ services=cls.services["lbrule"],
+ traffictype='Ingress'
+ )
+
+ NetworkACL.create(
+ cls.api_client,
+ networkid=cls.network_1.id,
+ services=cls.services["http_rule"],
+ traffictype='Egress'
+ )
+
@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)
@@ -827,11 +855,11 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
def tearDown(self):
try:
- #Clean up, terminate the created network offerings
+ # Clean up, terminate the created network offerings
cleanup_resources(self.api_client, self.cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
-
+
return
def validate_vpc_offering(self, vpc_offering):
@@ -839,22 +867,22 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
self.debug("Check if the VPC offering is created successfully?")
vpc_offs = VpcOffering.list(
- self.api_client,
- id=vpc_offering.id
- )
+ self.api_client,
+ id=vpc_offering.id
+ )
self.assertEqual(
- isinstance(vpc_offs, list),
- True,
- "List VPC offerings should return a valid list"
- )
+ isinstance(vpc_offs, list),
+ True,
+ "List VPC offerings should return a valid list"
+ )
self.assertEqual(
- vpc_offering.name,
- vpc_offs[0].name,
- "Name of the VPC offering should match with listVPCOff data"
- )
+ vpc_offering.name,
+ vpc_offs[0].name,
+ "Name of the VPC offering should match with listVPCOff data"
+ )
self.debug(
- "VPC offering is created successfully - %s" %
- vpc_offering.name)
+ "VPC offering is created successfully - %s" %
+ vpc_offering.name)
return
def validate_vpc_network(self, network, state=None):
@@ -862,25 +890,25 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
self.debug("Check if the VPC network is created successfully?")
vpc_networks = VPC.list(
- self.api_client,
- id=network.id
- )
+ self.api_client,
+ id=network.id
+ )
self.assertEqual(
- isinstance(vpc_networks, list),
- True,
- "List VPC network should return a valid list"
- )
+ isinstance(vpc_networks, list),
+ True,
+ "List VPC network should return a valid list"
+ )
self.assertEqual(
- network.name,
- vpc_networks[0].name,
- "Name of the VPC network should match with listVPC data"
- )
+ network.name,
+ vpc_networks[0].name,
+ "Name of the VPC network should match with listVPC data"
+ )
if state:
self.assertEqual(
- vpc_networks[0].state,
- state,
+ vpc_networks[0].state,
+ state,
"VPC state should be '%s'" % state
- )
+ )
self.debug("VPC network validated - %s" % network.name)
return
@@ -888,24 +916,25 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
""" Validate network rules
"""
vms = VirtualMachine.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
public_ips = PublicIPAddress.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
for vm, public_ip in zip(vms, public_ips):
try:
ssh_1 = vm.get_ssh_client(
- ipaddress=public_ip.ipaddress.ipaddress)
+ ipaddress=public_ip.ipaddress.ipaddress)
self.debug("SSH into VM is successfully")
- self.debug("Verifying if we can ping to outside world from VM?")
+ self.debug(
+ "Verifying if we can ping to outside world from VM?")
# Ping to outsite world
res = ssh_1.execute("ping -c 1 www.google.com")
# res = 64 bytes from maa03s17-in-f20.1e100.net (74.125.236.212):
@@ -915,14 +944,14 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
# rtt min/avg/max/mdev = 25.970/25.970/25.970/0.000 ms
except Exception as e:
self.fail("Failed to SSH into VM - %s, %s" %
- (public_ip.ipaddress.ipaddress, e))
+ (public_ip.ipaddress.ipaddress, e))
result = str(res)
self.assertEqual(
- result.count("1 received"),
- 1,
- "Ping to outside world from VM should be successful"
- )
+ result.count("1 received"),
+ 1,
+ "Ping to outside world from VM should be successful"
+ )
def migrate_router(self, router):
""" Migrate the router """
@@ -931,13 +960,13 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
hosts = Host.list(self.api_client, zoneid=self.zone.id, type='Routing')
self.assertEqual(
- isinstance(hosts, list),
- True,
- "List hosts should return a valid list"
- )
+ isinstance(hosts, list),
+ True,
+ "List hosts should return a valid list"
+ )
if len(hosts) < 2:
- raise unittest.SkipTest(
- "No host available for migration. Test requires atleast 2 hosts")
+ self.skipTest(
+ "No host available for migration. Test requires atleast 2 hosts")
# Remove the host of current VM from the hosts list
hosts[:] = [host for host in hosts if host.id != router.hostid]
@@ -945,13 +974,13 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
self.debug("Validating if the network rules work properly or not?")
self.debug("Migrating VM-ID: %s from %s to Host: %s" % (
- router.id,
- router.hostid,
- host.id
- ))
+ router.id,
+ router.hostid,
+ host.id
+ ))
try:
- #Migrate the router
+ # Migrate the router
cmd = migrateSystemVm.migrateSystemVmCmd()
cmd.isAsync = "false"
cmd.hostid = host.id
@@ -960,93 +989,95 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
except Exception as e:
self.fail("Failed to migrate instance, %s" % e)
-
+
self.debug("Waiting for Router mgiration ....")
- time.sleep(240)
-
- #List routers to check state of router
+ time.sleep(240)
+
+ # List routers to check state of router
router_response = list_routers(
- self.api_client,
- id=router.id
- )
+ self.api_client,
+ id=router.id
+ )
self.assertEqual(
- isinstance(router_response, list),
- True,
- "Check list response returns a valid list"
- )
-
+ isinstance(router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
+
router.hostid = router_response[0].hostid
- self.assertEqual(router.hostid, host.id, "Migration to host %s failed. The router host is"
- "still %s" % (host.id, router.hostid))
+ self.assertEqual(
+ router.hostid, host.id, "Migration to host %s failed. The router host is"
+ "still %s" %
+ (host.id, router.hostid))
return
-
@attr(tags=["advanced", "intervlan", "provisioining"])
def test_01_start_stop_router_after_addition_of_one_guest_network(self):
""" Test start/stop of router after addition of one guest network
- """
+ """
# Validations
- #1. Create a VPC with cidr - 10.1.1.1/16
- #2. Add network1(10.1.1.1/24) to this VPC.
- #3. Deploy vm1,vm2 and vm3 such that they are part of network1.
- #4. Create a PF /Static Nat/LB rule for vms in network1.
- #5. Create ingress network ACL for allowing all the above rules from a public ip range on network1.
- #6. Create egress network ACL for network1 to access google.com.
- #7. Create a private gateway for this VPC and add a static route to this gateway.
- #8. Create a VPN gateway for this VPC and add a static route to this gateway.
- #9. Make sure that all the PF,LB and Static NAT rules work as expected.
- #10. Make sure that we are able to access google.com from all the user Vms.
- #11. Make sure that the newly added private gateway's and VPN gateway's static routes work as expected
+ # 1. Create a VPC with cidr - 10.1.1.1/16
+ # 2. Add network1(10.1.1.1/24) to this VPC.
+ # 3. Deploy vm1,vm2 and vm3 such that they are part of network1.
+ # 4. Create a PF /Static Nat/LB rule for vms in network1.
+ # 5. Create ingress network ACL for allowing all the above rules from a public ip range on network1.
+ # 6. Create egress network ACL for network1 to access google.com.
+ # 7. Create a private gateway for this VPC and add a static route to this gateway.
+ # 8. Create a VPN gateway for this VPC and add a static route to this gateway.
+ # 9. Make sure that all the PF,LB and Static NAT rules work as expected.
+ # 10. Make sure that we are able to access google.com from all the user Vms.
+ # 11. Make sure that the newly added private gateway's and VPN
+ # gateway's static routes work as expected
self.validate_vpc_offering(self.vpc_off)
self.validate_vpc_network(self.vpc)
self.assertEqual(
- isinstance(self.gateways, list),
- True,
- "List private gateways should return a valid response"
- )
+ isinstance(self.gateways, list),
+ True,
+ "List private gateways should return a valid response"
+ )
self.assertEqual(
- isinstance(self.static_routes, list),
- True,
- "List static route should return a valid response"
- )
+ isinstance(self.static_routes, list),
+ True,
+ "List static route should return a valid response"
+ )
# Stop the VPC Router
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
- True,
- "List Routers should return a valid list"
- )
- router = routers[0]
+ isinstance(routers, list),
+ True,
+ "List Routers should return a valid list"
+ )
+ router = routers[0]
self.debug("Stopping the router with ID: %s" % router.id)
- #Stop the router
+ # Stop the router
cmd = stopRouter.stopRouterCmd()
cmd.id = router.id
self.api_client.stopRouter(cmd)
-
- #List routers to check state of router
+
+ # List routers to check state of router
router_response = list_routers(
- self.api_client,
- id=router.id
- )
+ self.api_client,
+ id=router.id
+ )
self.assertEqual(
- isinstance(router_response, list),
- True,
- "Check list response returns a valid list"
- )
- #List router should have router in stopped state
+ isinstance(router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
+ # List router should have router in stopped state
self.assertEqual(
- router_response[0].state,
- 'Stopped',
- "Check list router response for router state"
- )
+ router_response[0].state,
+ 'Stopped',
+ "Check list router response for router state"
+ )
self.debug("Stopped the router with ID: %s" % router.id)
@@ -1056,182 +1087,187 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
cmd.id = router.id
self.api_client.startRouter(cmd)
- #List routers to check state of router
+ # List routers to check state of router
router_response = list_routers(
- self.api_client,
- id=router.id
- )
+ self.api_client,
+ id=router.id
+ )
self.assertEqual(
- isinstance(router_response, list),
- True,
- "Check list response returns a valid list"
- )
- #List router should have router in running state
+ isinstance(router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
+ # List router should have router in running state
self.assertEqual(
- router_response[0].state,
- 'Running',
- "Check list router response for router state"
- )
+ router_response[0].state,
+ 'Running',
+ "Check list router response for router state"
+ )
self.debug("Started the router with ID: %s" % router.id)
-
+
return
@attr(tags=["advanced", "intervlan"], required_hardware="false")
def test_02_reboot_router_after_addition_of_one_guest_network(self):
""" Test reboot of router after addition of one guest network
- """
+ """
# Validations
- #1. Create a VPC with cidr - 10.1.1.1/16
- #2. Add network1(10.1.1.1/24) to this VPC.
- #3. Deploy vm1,vm2 and vm3 such that they are part of network1.
- #4. Create a PF /Static Nat/LB rule for vms in network1.
- #5. Create ingress network ACL for allowing all the above rules from a public ip range on network1.
- #6. Create egress network ACL for network1 to access google.com.
- #7. Create a private gateway for this VPC and add a static route to this gateway.
- #8. Create a VPN gateway for this VPC and add a static route to this gateway.
- #9. Make sure that all the PF,LB and Static NAT rules work as expected.
- #10. Make sure that we are able to access google.com from all the user Vms.
- #11. Make sure that the newly added private gateway's and VPN gateway's static routes work as expected
+ # 1. Create a VPC with cidr - 10.1.1.1/16
+ # 2. Add network1(10.1.1.1/24) to this VPC.
+ # 3. Deploy vm1,vm2 and vm3 such that they are part of network1.
+ # 4. Create a PF /Static Nat/LB rule for vms in network1.
+ # 5. Create ingress network ACL for allowing all the above rules from a public ip range on network1.
+ # 6. Create egress network ACL for network1 to access google.com.
+ # 7. Create a private gateway for this VPC and add a static route to this gateway.
+ # 8. Create a VPN gateway for this VPC and add a static route to this gateway.
+ # 9. Make sure that all the PF,LB and Static NAT rules work as expected.
+ # 10. Make sure that we are able to access google.com from all the user Vms.
+ # 11. Make sure that the newly added private gateway's and VPN
+ # gateway's static routes work as expected
self.validate_vpc_offering(self.vpc_off)
self.validate_vpc_network(self.vpc)
self.assertEqual(
- isinstance(self.gateways, list),
- True,
- "List private gateways should return a valid response"
- )
+ isinstance(self.gateways, list),
+ True,
+ "List private gateways should return a valid response"
+ )
self.assertEqual(
- isinstance(self.static_routes, list),
- True,
- "List static route should return a valid response"
- )
+ isinstance(self.static_routes, list),
+ True,
+ "List static route should return a valid response"
+ )
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
- True,
- "List Routers should return a valid list"
- )
- router = routers[0]
+ isinstance(routers, list),
+ True,
+ "List Routers should return a valid list"
+ )
+ router = routers[0]
self.debug("Rebooting the router ...")
- #Reboot the router
+ # Reboot the router
cmd = rebootRouter.rebootRouterCmd()
cmd.id = router.id
self.api_client.rebootRouter(cmd)
- #List routers to check state of router
+ # List routers to check state of router
router_response = list_routers(
- self.api_client,
- id=router.id
- )
+ self.api_client,
+ id=router.id
+ )
self.assertEqual(
- isinstance(router_response, list),
- True,
- "Check list response returns a valid list"
- )
- #List router should have router in running state and same public IP
+ isinstance(router_response, list),
+ True,
+ "Check list response returns a valid list"
+ )
+ # List router should have router in running state and same public IP
self.assertEqual(
- router_response[0].state,
- 'Running',
- "Check list router response for router state"
- )
+ router_response[0].state,
+ 'Running',
+ "Check list router response for router state"
+ )
return
@attr(tags=["advanced", "intervlan"], required_hardware="true")
def test_03_migrate_router_after_addition_of_one_guest_network(self):
""" Test migrate of router after addition of one guest network
- """
+ """
# Validations
- #1. Create a VPC with cidr - 10.1.1.1/16
- #2. Add network1(10.1.1.1/24) to this VPC.
- #3. Deploy vm1,vm2 and vm3 such that they are part of network1.
- #4. Create a PF /Static Nat/LB rule for vms in network1.
- #5. Create ingress network ACL for allowing all the above rules from a public ip range on network1.
- #6. Create egress network ACL for network1 to access google.com.
- #7. Create a private gateway for this VPC and add a static route to this gateway.
- #8. Create a VPN gateway for this VPC and add a static route to this gateway.
- #9. Make sure that all the PF,LB and Static NAT rules work as expected.
- #10. Make sure that we are able to access google.com from all the user Vms.
- #11. Make sure that the newly added private gateway's and VPN gateway's static routes work as expected
+ # 1. Create a VPC with cidr - 10.1.1.1/16
+ # 2. Add network1(10.1.1.1/24) to this VPC.
+ # 3. Deploy vm1,vm2 and vm3 such that they are part of network1.
+ # 4. Create a PF /Static Nat/LB rule for vms in network1.
+ # 5. Create ingress network ACL for allowing all the above rules from a public ip range on network1.
+ # 6. Create egress network ACL for network1 to access google.com.
+ # 7. Create a private gateway for this VPC and add a static route to this gateway.
+ # 8. Create a VPN gateway for this VPC and add a static route to this gateway.
+ # 9. Make sure that all the PF,LB and Static NAT rules work as expected.
+ # 10. Make sure that we are able to access google.com from all the user Vms.
+ # 11. Make sure that the newly added private gateway's and VPN
+ # gateway's static routes work as expected
self.hypervisor = self.testClient.getHypervisorInfo()
if self.hypervisor.lower() in ['lxc']:
- self.skipTest("vm migrate is not supported in %s" % self.hypervisor)
+ self.skipTest(
+ "vm migrate is not supported in %s" %
+ self.hypervisor)
self.validate_vpc_offering(self.vpc_off)
self.validate_vpc_network(self.vpc)
self.assertEqual(
- isinstance(self.gateways, list),
- True,
- "List private gateways should return a valid response"
- )
+ isinstance(self.gateways, list),
+ True,
+ "List private gateways should return a valid response"
+ )
self.assertEqual(
- isinstance(self.static_routes, list),
- True,
- "List static route should return a valid response"
- )
+ isinstance(self.static_routes, list),
+ True,
+ "List static route should return a valid response"
+ )
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
- True,
- "List Routers should return a valid list"
- )
+ isinstance(routers, list),
+ True,
+ "List Routers should return a valid list"
+ )
self.migrate_router(routers[0])
return
@attr(tags=["advanced", "intervlan"], required_hardware="false")
def test_04_chg_srv_off_router_after_addition_of_one_guest_network(self):
""" Test to change service offering of router after addition of one guest network
- """
+ """
# Validations
- #1. Create a VPC with cidr - 10.1.1.1/16
- #2. Add network1(10.1.1.1/24) to this VPC.
- #3. Deploy vm1,vm2 and vm3 such that they are part of network1.
- #4. Create a PF /Static Nat/LB rule for vms in network1.
- #5. Create ingress network ACL for allowing all the above rules from a public ip range on network1.
- #6. Create egress network ACL for network1 to access google.com.
- #7. Create a private gateway for this VPC and add a static route to this gateway.
- #8. Create a VPN gateway for this VPC and add a static route to this gateway.
- #9. Make sure that all the PF,LB and Static NAT rules work as expected.
- #10. Make sure that we are able to access google.com from all the user Vms.
- #11. Make sure that the newly added private gateway's and VPN gateway's static routes work as expected
+ # 1. Create a VPC with cidr - 10.1.1.1/16
+ # 2. Add network1(10.1.1.1/24) to this VPC.
+ # 3. Deploy vm1,vm2 and vm3 such that they are part of network1.
+ # 4. Create a PF /Static Nat/LB rule for vms in network1.
+ # 5. Create ingress network ACL for allowing all the above rules from a public ip range on network1.
+ # 6. Create egress network ACL for network1 to access google.com.
+ # 7. Create a private gateway for this VPC and add a static route to this gateway.
+ # 8. Create a VPN gateway for this VPC and add a static route to this gateway.
+ # 9. Make sure that all the PF,LB and Static NAT rules work as expected.
+ # 10. Make sure that we are able to access google.com from all the user Vms.
+ # 11. Make sure that the newly added private gateway's and VPN
+ # gateway's static routes work as expected
self.validate_vpc_offering(self.vpc_off)
self.validate_vpc_network(self.vpc)
self.assertEqual(
- isinstance(self.gateways, list),
- True,
- "List private gateways should return a valid response"
- )
+ isinstance(self.gateways, list),
+ True,
+ "List private gateways should return a valid response"
+ )
self.assertEqual(
- isinstance(self.static_routes, list),
- True,
- "List static route should return a valid response"
- )
-
+ isinstance(self.static_routes, list),
+ True,
+ "List static route should return a valid response"
+ )
+
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
- True,
- "List Routers should return a valid list"
- )
-
- #Stop the router
+ isinstance(routers, list),
+ True,
+ "List Routers should return a valid list"
+ )
+
+ # Stop the router
router = routers[0]
self.debug("Stopping the router with ID: %s" % router.id)
cmd = stopRouter.stopRouterCmd()
@@ -1239,32 +1275,32 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
self.api_client.stopRouter(cmd)
service_offering = ServiceOffering.create(
- self.api_client,
- self.services["service_offering_new"]
- )
+ self.api_client,
+ self.services["service_offering_new"]
+ )
self.debug("Changing service offering for the Router %s" % router.id)
- try:
+ try:
router = Router.change_service_offering(self.api_client,
- router.id,
- service_offering.id
- )
+ router.id,
+ service_offering.id
+ )
except:
self.fail("Changing service offering failed")
-
- self.debug("Router %s" % router)
+
+ self.debug("Router %s" % router)
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
router = routers[0]
self.assertEqual(
- router.serviceofferingid,
- service_offering.id,
- "Changing service offering failed as id is %s and expected"
- "is %s" % (router.serviceofferingid, service_offering.id)
- )
+ router.serviceofferingid,
+ service_offering.id,
+ "Changing service offering failed as id is %s and expected"
+ "is %s" % (router.serviceofferingid, service_offering.id)
+ )
return
@attr(tags=["advanced", "intervlan"], required_hardware="false")
@@ -1272,56 +1308,57 @@ class TestVPCRouterOneNetwork(cloudstackTestCase):
""" Test destroy of router after addition of one guest network
"""
# Validations
- #1. Create a VPC with cidr - 10.1.1.1/16
- #2. Add network1(10.1.1.1/24) to this VPC.
- #3. Deploy vm1,vm2 and vm3 such that they are part of network1.
- #4. Create a PF /Static Nat/LB rule for vms in network1.
- #5. Create ingress network ACL for allowing all the above rules from a public ip range on network1.
- #6. Create egress network ACL for network1 to access google.com.
- #7. Create a private gateway for this VPC and add a static route to this gateway.
- #8. Create a VPN gateway for this VPC and add a static route to this gateway.
- #9. Make sure that all the PF,LB and Static NAT rules work as expected.
- #10. Make sure that we are able to access google.com from all the user Vms.
- #11. Make sure that the newly added private gateway's and VPN gateway's static routes work as expected
+ # 1. Create a VPC with cidr - 10.1.1.1/16
+ # 2. Add network1(10.1.1.1/24) to this VPC.
+ # 3. Deploy vm1,vm2 and vm3 such that they are part of network1.
+ # 4. Create a PF /Static Nat/LB rule for vms in network1.
+ # 5. Create ingress network ACL for allowing all the above rules from a public ip range on network1.
+ # 6. Create egress network ACL for network1 to access google.com.
+ # 7. Create a private gateway for this VPC and add a static route to this gateway.
+ # 8. Create a VPN gateway for this VPC and add a static route to this gateway.
+ # 9. Make sure that all the PF,LB and Static NAT rules work as expected.
+ # 10. Make sure that we are able to access google.com from all the user Vms.
+ # 11. Make sure that the newly added private gateway's and VPN
+ # gateway's static routes work as expected
self.validate_vpc_offering(self.vpc_off)
self.validate_vpc_network(self.vpc)
self.assertEqual(
- isinstance(self.gateways, list),
- True,
- "List private gateways should return a valid response"
- )
+ isinstance(self.gateways, list),
+ True,
+ "List private gateways should return a valid response"
+ )
self.assertEqual(
- isinstance(self.static_routes, list),
- True,
- "List static route should return a valid response"
- )
+ isinstance(self.static_routes, list),
+ True,
+ "List static route should return a valid response"
+ )
routers = Router.list(
- self.api_client,
- account=self.account.name,
- domainid=self.account.domainid,
- listall=True
- )
+ self.api_client,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ listall=True
+ )
self.assertEqual(
- isinstance(routers, list),
- True,
-
<TRUNCATED>
[2/4] git commit: updated refs/heads/master to 6c3dc15
Posted by ga...@apache.org.
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/6c3dc15f/test/integration/component/test_snapshot_limits.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_snapshot_limits.py b/test/integration/component/test_snapshot_limits.py
index 6d8c980..21e15da 100644
--- a/test/integration/component/test_snapshot_limits.py
+++ b/test/integration/component/test_snapshot_limits.py
@@ -17,7 +17,6 @@
from nose.plugins.attrib import attr
from marvin.cloudstackTestCase import cloudstackTestCase, unittest
-#from marvin.cloudstackAPI import *
from marvin.lib.utils import cleanup_resources
from marvin.lib.base import (Account,
VirtualMachine,
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/6c3dc15f/test/integration/component/test_templates.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_templates.py b/test/integration/component/test_templates.py
index eafc1bb..7973766 100644
--- a/test/integration/component/test_templates.py
+++ b/test/integration/component/test_templates.py
@@ -16,7 +16,7 @@
# under the License.
""" P1 tests for Templates
"""
-#Import Local Modules
+# Import Local Modules
from nose.plugins.attrib import attr
from marvin.cloudstackTestCase import cloudstackTestCase, unittest
from marvin.cloudstackAPI import listZones
@@ -31,78 +31,79 @@ from marvin.lib.common import (get_domain,
get_zone,
get_template,
get_builtin_template_info)
-#Import System modules
+# Import System modules
import time
class Services:
+
"""Test Templates Services
"""
def __init__(self):
self.services = {
- "account": {
- "email": "test@test.com",
- "firstname": "Test",
- "lastname": "User",
- "username": "test",
- # Random characters are appended for unique
- # username
- "password": "password",
- },
- "service_offering": {
- "name": "Tiny Instance",
- "displaytext": "Tiny Instance",
- "cpunumber": 1,
- "cpuspeed": 100, # in MHz
+ "account": {
+ "email": "test@test.com",
+ "firstname": "Test",
+ "lastname": "User",
+ "username": "test",
+ # Random characters are appended for unique
+ # username
+ "password": "password",
+ },
+ "service_offering": {
+ "name": "Tiny Instance",
+ "displaytext": "Tiny Instance",
+ "cpunumber": 1,
+ "cpuspeed": 100, # in MHz
"memory": 128, # In MBs
- },
- "disk_offering": {
- "displaytext": "Small",
- "name": "Small",
- "disksize": 1
- },
- "virtual_machine": {
- "displayname": "testVM",
- "hypervisor": 'XenServer',
- "protocol": 'TCP',
- "ssh_port": 22,
- "username": "root",
- "password": "password",
- "privateport": 22,
- "publicport": 22,
- },
- "volume": {
- "diskname": "Test Volume",
- },
- "templates": {
- # Configs for different Template formats
- # For Eg. raw image, zip etc
- 0: {
- "displaytext": "Public Template",
- "name": "Public template",
- "ostype": 'CentOS 5.3 (64-bit)',
- "url": "http://download.cloud.com/releases/2.0.0/UbuntuServer-10-04-64bit.vhd.bz2",
- "hypervisor": 'XenServer',
- "format": 'VHD',
- "isfeatured": True,
- "ispublic": True,
- "isextractable": True,
- },
- },
- "template": {
- "displaytext": "Cent OS Template",
- "name": "Cent OS Template",
- "ostype": 'CentOS 5.3 (64-bit)',
- "templatefilter": 'self',
- "isfeatured": True,
- "ispublic": True,
- },
- "templatefilter": 'self',
- "ostype": 'CentOS 5.3 (64-bit)',
- "sleep": 60,
- "timeout": 10,
- }
+ },
+ "disk_offering": {
+ "displaytext": "Small",
+ "name": "Small",
+ "disksize": 1
+ },
+ "virtual_machine": {
+ "displayname": "testVM",
+ "hypervisor": 'XenServer',
+ "protocol": 'TCP',
+ "ssh_port": 22,
+ "username": "root",
+ "password": "password",
+ "privateport": 22,
+ "publicport": 22,
+ },
+ "volume": {
+ "diskname": "Test Volume",
+ },
+ "templates": {
+ # Configs for different Template formats
+ # For Eg. raw image, zip etc
+ 0: {
+ "displaytext": "Public Template",
+ "name": "Public template",
+ "ostype": 'CentOS 5.3 (64-bit)',
+ "url": "http://download.cloud.com/releases/2.0.0/UbuntuServer-10-04-64bit.vhd.bz2",
+ "hypervisor": 'XenServer',
+ "format": 'VHD',
+ "isfeatured": True,
+ "ispublic": True,
+ "isextractable": True,
+ },
+ },
+ "template": {
+ "displaytext": "Cent OS Template",
+ "name": "Cent OS Template",
+ "ostype": 'CentOS 5.3 (64-bit)',
+ "templatefilter": 'self',
+ "isfeatured": True,
+ "ispublic": True,
+ },
+ "templatefilter": 'self',
+ "ostype": 'CentOS 5.3 (64-bit)',
+ "sleep": 60,
+ "timeout": 10,
+ }
class TestCreateTemplate(cloudstackTestCase):
@@ -111,11 +112,15 @@ class TestCreateTemplate(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.unsupportedHypervisor:
+ self.skipTest(
+ "Template creation from root volume is not supported in LXC")
return
def tearDown(self):
try:
- #Clean up, terminate the created templates
+ # Clean up, terminate the created templates
cleanup_resources(self.apiclient, self.cleanup)
except Exception as e:
@@ -132,33 +137,35 @@ class TestCreateTemplate(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.unsupportedHypervisor = False
cls.hypervisor = cls.testClient.getHypervisorInfo()
if cls.hypervisor.lower() in ['lxc']:
- raise unittest.SkipTest("Template creation from root volume is not supported in LXC")
+ cls.unsupportedHypervisor = True
+ return
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
cls.service_offering = ServiceOffering.create(
- cls.api_client,
- cls.services["service_offering"]
- )
+ cls.api_client,
+ cls.services["service_offering"]
+ )
+ cls._cleanup.append(cls.service_offering)
cls.account = Account.create(
- cls.api_client,
- cls.services["account"],
- domainid=cls.domain.id
- )
+ cls.api_client,
+ cls.services["account"],
+ domainid=cls.domain.id
+ )
+ cls._cleanup.append(cls.account)
cls.services["account"] = cls.account.name
-
- cls._cleanup = [
- cls.account,
- cls.service_offering
- ]
return
@classmethod
def tearDownClass(cls):
try:
- cls.api_client = super(TestCreateTemplate, cls).getClsTestClient().getApiClient()
- #Cleanup resources used
+ cls.api_client = super(
+ TestCreateTemplate,
+ cls).getClsTestClient().getApiClient()
+ # Cleanup resources used
cleanup_resources(cls.api_client, cls._cleanup)
except Exception as e:
@@ -190,18 +197,18 @@ class TestCreateTemplate(cloudstackTestCase):
# Register new template
template = Template.register(
- self.apiclient,
- self.services["templates"][0],
- zoneid=self.zone.id,
- account=self.account.name,
- domainid=self.account.domainid,
- hypervisor=self.hypervisor
- )
+ self.apiclient,
+ self.services["templates"][0],
+ zoneid=self.zone.id,
+ account=self.account.name,
+ domainid=self.account.domainid,
+ hypervisor=self.hypervisor
+ )
self.debug(
- "Registered a template of format: %s with ID: %s" % (
- self.services["templates"][0]["format"],
- template.id
- ))
+ "Registered a template of format: %s with ID: %s" % (
+ self.services["templates"][0]["format"],
+ template.id
+ ))
# Wait for template to download
template.download(self.apiclient)
self.cleanup.append(template)
@@ -211,12 +218,12 @@ class TestCreateTemplate(cloudstackTestCase):
timeout = self.services["timeout"]
while True:
list_template_response = Template.list(
- self.apiclient,
- templatefilter='all',
- id=template.id,
- zoneid=self.zone.id,
- account=self.account.name,
- domainid=self.account.domainid)
+ self.apiclient,
+ templatefilter='all',
+ id=template.id,
+ zoneid=self.zone.id,
+ account=self.account.name,
+ domainid=self.account.domainid)
if isinstance(list_template_response, list):
break
elif timeout == 0:
@@ -224,58 +231,58 @@ class TestCreateTemplate(cloudstackTestCase):
time.sleep(5)
timeout = timeout - 1
- #Verify template response to check whether template added successfully
+ # Verify template response to check whether template added successfully
self.assertEqual(
- isinstance(list_template_response, list),
- True,
- "Check for list template response return valid data"
- )
+ isinstance(list_template_response, list),
+ True,
+ "Check for list template response return valid data"
+ )
self.assertNotEqual(
- len(list_template_response),
- 0,
- "Check template available in List Templates"
- )
+ len(list_template_response),
+ 0,
+ "Check template available in List Templates"
+ )
template_response = list_template_response[0]
self.assertEqual(
- template_response.isready,
- True,
- "Template state is not ready, it is %s" % template_response.isready
- )
+ template_response.isready,
+ True,
+ "Template state is not ready, it is %s" % template_response.isready
+ )
# Deploy new virtual machine using template
virtual_machine = VirtualMachine.create(
- self.apiclient,
- self.services["virtual_machine"],
- templateid=template.id,
- accountid=self.account.name,
- domainid=self.account.domainid,
- serviceofferingid=self.service_offering.id,
- mode=self.services["mode"]
- )
+ self.apiclient,
+ self.services["virtual_machine"],
+ templateid=template.id,
+ accountid=self.account.name,
+ domainid=self.account.domainid,
+ serviceofferingid=self.service_offering.id,
+ mode=self.services["mode"]
+ )
self.debug("creating an instance with template ID: %s" % template.id)
vm_response = VirtualMachine.list(self.apiclient,
id=virtual_machine.id,
account=self.account.name,
domainid=self.account.domainid)
self.assertEqual(
- isinstance(vm_response, list),
- True,
- "Check for list VMs response after VM deployment"
- )
- #Verify VM response to check whether VM deployment was successful
+ isinstance(vm_response, list),
+ True,
+ "Check for list VMs response after VM deployment"
+ )
+ # Verify VM response to check whether VM deployment was successful
self.assertNotEqual(
- len(vm_response),
- 0,
- "Check VMs available in List VMs response"
- )
+ len(vm_response),
+ 0,
+ "Check VMs available in List VMs response"
+ )
vm = vm_response[0]
self.assertEqual(
- vm.state,
- 'Running',
- "Check the state of VM created from Template"
- )
+ vm.state,
+ 'Running',
+ "Check the state of VM created from Template"
+ )
return
@@ -286,16 +293,18 @@ class TestTemplates(cloudstackTestCase):
cls.testClient = super(TestTemplates, cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
- cls.hypervisor = cls.testClient.getHypervisorInfo()
cls.services = Services().services
# 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._cleanup = []
+ cls.unsupportedHypervisor = False
cls.hypervisor = cls.testClient.getHypervisorInfo()
if cls.hypervisor.lower() in ['lxc']:
- raise unittest.SkipTest("Template creation from root volume is not supported in LXC")
- #populate second zone id for iso copy
+ cls.unsupportedHypervisor = True
+ return
+ # populate second zone id for iso copy
cmd = listZones.listZonesCmd()
zones = cls.api_client.listZones(cmd)
if not isinstance(zones, list):
@@ -304,47 +313,46 @@ class TestTemplates(cloudstackTestCase):
cls.services["destzoneid"] = zones[1].id
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._cleanup = []
try:
cls.account = Account.create(
- cls.api_client,
- cls.services["account"],
- domainid=cls.domain.id
- )
+ cls.api_client,
+ cls.services["account"],
+ domainid=cls.domain.id
+ )
cls._cleanup.append(cls.account)
cls.services["account"] = cls.account.name
cls.service_offering = ServiceOffering.create(
- cls.api_client,
- cls.services["service_offering"]
- )
+ cls.api_client,
+ cls.services["service_offering"]
+ )
cls._cleanup.append(cls.service_offering)
# create virtual machine
cls.virtual_machine = VirtualMachine.create(
- cls.api_client,
- cls.services["virtual_machine"],
- templateid=template.id,
- accountid=cls.account.name,
- domainid=cls.account.domainid,
- serviceofferingid=cls.service_offering.id,
- )
- #Stop virtual machine
+ cls.api_client,
+ cls.services["virtual_machine"],
+ templateid=template.id,
+ accountid=cls.account.name,
+ domainid=cls.account.domainid,
+ serviceofferingid=cls.service_offering.id,
+ )
+ # Stop virtual machine
cls.virtual_machine.stop(cls.api_client)
timeout = cls.services["timeout"]
while True:
list_volume = Volume.list(
- cls.api_client,
- virtualmachineid=cls.virtual_machine.id,
- type='ROOT',
- listall=True)
+ cls.api_client,
+ virtualmachineid=cls.virtual_machine.id,
+ type='ROOT',
+ listall=True)
if isinstance(list_volume, list):
break
elif timeout == 0:
@@ -355,12 +363,12 @@ class TestTemplates(cloudstackTestCase):
cls.volume = list_volume[0]
- #Create template from volume
+ # Create template from volume
cls.template = Template.create(
- cls.api_client,
- cls.services["template"],
- cls.volume.id
- )
+ cls.api_client,
+ cls.services["template"],
+ cls.volume.id
+ )
except Exception as e:
cls.tearDownClass()
raise unittest.SkipTest("Failure in setUpClass: %s" % e)
@@ -368,8 +376,10 @@ class TestTemplates(cloudstackTestCase):
@classmethod
def tearDownClass(cls):
try:
- cls.api_client = super(TestTemplates, cls).getClsTestClient().getApiClient()
- #Cleanup created resources such as templates and VMs
+ cls.api_client = super(
+ TestTemplates,
+ cls).getClsTestClient().getApiClient()
+ # Cleanup created resources such as templates and VMs
cleanup_resources(cls.api_client, cls._cleanup)
except Exception as e:
@@ -382,11 +392,15 @@ class TestTemplates(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.unsupportedHypervisor:
+ self.skipTest(
+ "Template creation from root volume is not supported in LXC")
return
def tearDown(self):
try:
- #Clean up, terminate the created templates
+ # Clean up, terminate the created templates
cleanup_resources(self.apiclient, self.cleanup)
except Exception as e:
@@ -404,33 +418,35 @@ class TestTemplates(cloudstackTestCase):
# 2. VM should be in Up and Running state
virtual_machine = VirtualMachine.create(
- self.apiclient,
- self.services["virtual_machine"],
- templateid=self.template.id,
- accountid=self.account.name,
- domainid=self.account.domainid,
- serviceofferingid=self.service_offering.id,
- )
-
- self.debug("creating an instance with template ID: %s" % self.template.id)
+ self.apiclient,
+ self.services["virtual_machine"],
+ templateid=self.template.id,
+ accountid=self.account.name,
+ domainid=self.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 = VirtualMachine.list(
- self.apiclient,
- id=virtual_machine.id,
- account=self.account.name,
- domainid=self.account.domainid)
- #Verify VM response to check whether VM deployment was successful
+ self.apiclient,
+ id=virtual_machine.id,
+ account=self.account.name,
+ domainid=self.account.domainid)
+ # Verify VM response to check whether VM deployment was successful
self.assertNotEqual(
- len(vm_response),
- 0,
- "Check VMs available in List VMs response"
- )
+ len(vm_response),
+ 0,
+ "Check VMs available in List VMs response"
+ )
vm = vm_response[0]
self.assertEqual(
- vm.state,
- 'Running',
- "Check the state of VM created from Template"
- )
+ vm.state,
+ 'Running',
+ "Check the state of VM created from Template"
+ )
return
@attr(tags=["advanced", "advancedns"], required_hardware="false")
@@ -440,34 +456,34 @@ class TestTemplates(cloudstackTestCase):
# Validate the following:
# 1. Create a template and verify it is shown in list templates response
- # 2. Delete the created template and again verify list template response
+ # 2. Delete the created template and again verify list template
+ # response
# Verify template response for updated attributes
list_template_response = Template.list(
- self.apiclient,
- templatefilter=\
- self.services["template"]["templatefilter"],
- id=self.template.id,
- zoneid=self.zone.id)
+ self.apiclient,
+ templatefilter=self.services["template"]["templatefilter"],
+ id=self.template.id,
+ zoneid=self.zone.id)
self.assertEqual(
- isinstance(list_template_response, list),
- True,
- "Check for list template response return valid list"
- )
+ isinstance(list_template_response, list),
+ True,
+ "Check for list template response return valid list"
+ )
self.assertNotEqual(
- len(list_template_response),
- 0,
- "Check template available in List Templates"
- )
+ len(list_template_response),
+ 0,
+ "Check template available in List Templates"
+ )
template_response = list_template_response[0]
self.assertEqual(
- template_response.id,
- self.template.id,
- "Template id %s in the list is not matching with created template id %s" %
- (template_response.id, self.template.id)
- )
+ template_response.id,
+ self.template.id,
+ "Template id %s in the list is not matching with created template id %s" %
+ (template_response.id, self.template.id)
+ )
self.debug("Deleting template: %s" % self.template)
# Delete the template
@@ -475,20 +491,19 @@ class TestTemplates(cloudstackTestCase):
self.debug("Delete template: %s successful" % self.template)
list_template_response = Template.list(
- self.apiclient,
- templatefilter=\
- self.services["template"]["templatefilter"],
- id=self.template.id,
- zoneid=self.zone.id
- )
+ self.apiclient,
+ templatefilter=self.services["template"]["templatefilter"],
+ id=self.template.id,
+ zoneid=self.zone.id
+ )
self.assertEqual(
- list_template_response,
- None,
- "Check template available in List Templates"
- )
+ list_template_response,
+ None,
+ "Check template available in List Templates"
+ )
return
- @attr(speed = "slow")
+ @attr(speed="slow")
@attr(tags=["advanced", "advancedns"], required_hardware="false")
def test_04_template_from_snapshot(self):
"""Create Template from snapshot
@@ -501,88 +516,89 @@ class TestTemplates(cloudstackTestCase):
# 5. VM should be in running state
if self.hypervisor.lower() in ['hyperv', 'lxc']:
- self.skipTest("Snapshots feature is not supported on %s" % self.hypervisor.lower())
+ self.skipTest(
+ "Snapshots feature is not supported on %s" %
+ self.hypervisor.lower())
userapiclient = self.testClient.getUserApiClient(
- UserName=self.account.name,
- DomainName=self.account.domain)
+ UserName=self.account.name,
+ DomainName=self.account.domain)
volumes = Volume.list(
- userapiclient,
- virtualmachineid=self.virtual_machine.id,
- type='ROOT',
- listall=True
- )
+ userapiclient,
+ virtualmachineid=self.virtual_machine.id,
+ type='ROOT',
+ listall=True
+ )
volume = volumes[0]
self.debug("Creating a snapshot from volume: %s" % volume.id)
- #Create a snapshot of volume
+ # Create a snapshot of volume
snapshot = Snapshot.create(
- userapiclient,
- volume.id,
- account=self.account.name,
- domainid=self.account.domainid
- )
+ userapiclient,
+ volume.id,
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
self.debug("Creating a template from snapshot: %s" % snapshot.id)
# Generate template from the snapshot
template = Template.create_from_snapshot(
- userapiclient,
- snapshot,
- self.services["template"]
- )
+ userapiclient,
+ snapshot,
+ self.services["template"]
+ )
self.cleanup.append(template)
# Verify created template
templates = Template.list(
- userapiclient,
- templatefilter=\
- self.services["template"]["templatefilter"],
- id=template.id
- )
+ userapiclient,
+ templatefilter=self.services["template"]["templatefilter"],
+ id=template.id
+ )
self.assertNotEqual(
- templates,
- None,
- "Check if result exists in list item call"
- )
+ templates,
+ None,
+ "Check if result exists in list item call"
+ )
self.assertEqual(
- templates[0].id,
- template.id,
- "Check new template id in list resources call"
- )
+ templates[0].id,
+ template.id,
+ "Check new template id in list resources call"
+ )
self.debug("Deploying a VM from template: %s" % template.id)
# Deploy new virtual machine using template
virtual_machine = VirtualMachine.create(
- userapiclient,
- self.services["virtual_machine"],
- templateid=template.id,
- accountid=self.account.name,
- domainid=self.account.domainid,
- serviceofferingid=self.service_offering.id,
- )
+ userapiclient,
+ self.services["virtual_machine"],
+ templateid=template.id,
+ accountid=self.account.name,
+ domainid=self.account.domainid,
+ serviceofferingid=self.service_offering.id,
+ )
self.cleanup.append(virtual_machine)
vm_response = VirtualMachine.list(
- userapiclient,
- id=virtual_machine.id,
- account=self.account.name,
- domainid=self.account.domainid
- )
+ userapiclient,
+ id=virtual_machine.id,
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
self.assertEqual(
- isinstance(vm_response, list),
- True,
- "Check for list VM response return valid list"
- )
+ isinstance(vm_response, list),
+ True,
+ "Check for list VM response return valid list"
+ )
- #Verify VM response to check whether VM deployment was successful
+ # Verify VM response to check whether VM deployment was successful
self.assertNotEqual(
- len(vm_response),
- 0,
- "Check VMs available in List VMs response"
- )
+ len(vm_response),
+ 0,
+ "Check VMs available in List VMs response"
+ )
vm = vm_response[0]
self.assertEqual(
- vm.state,
- 'Running',
- "Check the state of VM created from Template"
- )
+ vm.state,
+ 'Running',
+ "Check the state of VM created from Template"
+ )
return
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/6c3dc15f/test/integration/component/test_usage.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_usage.py b/test/integration/component/test_usage.py
index c0d4031..5eb4c74 100644
--- a/test/integration/component/test_usage.py
+++ b/test/integration/component/test_usage.py
@@ -493,14 +493,18 @@ class TestVolumeUsage(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.unsupportedStorageType = False
cls.hypervisor = cls.testClient.getHypervisorInfo()
if cls.hypervisor.lower() == 'lxc':
if not find_storage_pool_type(cls.api_client, storagetype='rbd'):
- raise unittest.SkipTest("RBD storage type is required for data volumes for LXC")
+ cls.unsupportedStorageType = True
+ return
cls.disk_offering = DiskOffering.create(
cls.api_client,
cls.services["disk_offering"]
)
+ cls._cleanup.append(cls.disk_offering)
template = get_template(
cls.api_client,
cls.zone.id,
@@ -516,6 +520,7 @@ class TestVolumeUsage(cloudstackTestCase):
cls.services["account"],
domainid=cls.domain.id
)
+ cls._cleanup.append(cls.account)
cls.services["account"] = cls.account.name
@@ -523,6 +528,7 @@ class TestVolumeUsage(cloudstackTestCase):
cls.api_client,
cls.services["service_offering"]
)
+ cls._cleanup.append(cls.service_offering)
cls.virtual_machine = VirtualMachine.create(
cls.api_client,
cls.services["server"],
@@ -531,11 +537,6 @@ class TestVolumeUsage(cloudstackTestCase):
domainid=cls.account.domainid,
serviceofferingid=cls.service_offering.id
)
- cls._cleanup = [
- cls.service_offering,
- cls.disk_offering,
- cls.account,
- ]
return
@classmethod
@@ -551,6 +552,9 @@ class TestVolumeUsage(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.unsupportedStorageType:
+ self.skipTest("Skipping because of unsupported storage type")
return
def tearDown(self):
@@ -1177,14 +1181,17 @@ class TestSnapshotUsage(cloudstackTestCase):
def setUpClass(cls):
cls.testClient = super(TestSnapshotUsage, cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
- cls.hypervisor = cls.testClient.getHypervisorInfo()
cls.services = Services().services
# 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
- if cls.hypervisor.lower() == 'lxc':
- raise unittest.SkipTest("snapshots are not supported on LXC")
+ cls._cleanup = []
+ cls.unsupportedHypervisor = False
+ cls.hypervisor = cls.testClient.getHypervisorInfo()
+ if cls.hypervisor.lower() in ['hyperv', 'lxc']:
+ cls.unsupportedHypervisor = True
+ return
template = get_template(
cls.api_client,
@@ -1235,6 +1242,10 @@ class TestSnapshotUsage(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.unsupportedHypervisor:
+ self.skipTest("Snapshots are not supported on %s" %
+ self.hypervisor)
return
def tearDown(self):
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/6c3dc15f/test/integration/component/test_volumes.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_volumes.py b/test/integration/component/test_volumes.py
index 1b01e77..dd8b5e8 100644
--- a/test/integration/component/test_volumes.py
+++ b/test/integration/component/test_volumes.py
@@ -18,7 +18,7 @@
"""
# Import Local Modules
from nose.plugins.attrib import attr
-from marvin.cloudstackTestCase import cloudstackTestCase, unittest
+from marvin.cloudstackTestCase import cloudstackTestCase
from marvin.cloudstackAPI import (listHypervisorCapabilities,
attachIso,
deleteVolume)
@@ -114,14 +114,18 @@ class TestAttachVolume(cloudstackTestCase):
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.pod = get_pod(cls.api_client, cls.zone.id)
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.unsupportedStorageType = False
cls.hypervisor = cls.testClient.getHypervisorInfo()
if cls.hypervisor.lower() == 'lxc':
if not find_storage_pool_type(cls.api_client, storagetype='rbd'):
- raise unittest.SkipTest("RBD storage type is required for data volumes for LXC")
+ cls.unsupportedStorageType = True
+ return
cls.disk_offering = DiskOffering.create(
cls.api_client,
cls.services["disk_offering"]
)
+ cls._cleanup.append(cls.disk_offering)
template = get_template(
cls.api_client,
cls.zone.id,
@@ -155,11 +159,13 @@ class TestAttachVolume(cloudstackTestCase):
cls.services["account"],
domainid=cls.domain.id
)
+ cls._cleanup.append(cls.account)
cls.service_offering = ServiceOffering.create(
cls.api_client,
cls.services["service_offering"]
)
+ cls._cleanup.append(cls.service_offering)
cls.virtual_machine = VirtualMachine.create(
cls.api_client,
cls.services["virtual_machine"],
@@ -167,11 +173,6 @@ class TestAttachVolume(cloudstackTestCase):
domainid=cls.account.domainid,
serviceofferingid=cls.service_offering.id,
)
- cls._cleanup = [
- cls.service_offering,
- cls.disk_offering,
- cls.account
- ]
def setUp(self):
@@ -179,6 +180,9 @@ class TestAttachVolume(cloudstackTestCase):
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+ if self.unsupportedStorageType:
+ self.skipTest("Skipping because of unsupported storage type")
+
def tearDown(self):
try:
cleanup_resources(self.apiclient, self.cleanup)
@@ -379,14 +383,18 @@ class TestAttachDetachVolume(cloudstackTestCase):
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.pod = get_pod(cls.api_client, cls.zone.id)
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.unsupportedStorageType = False
cls.hypervisor = cls.testClient.getHypervisorInfo()
if cls.hypervisor.lower() == 'lxc':
if not find_storage_pool_type(cls.api_client, storagetype='rbd'):
- raise unittest.SkipTest("RBD storage type is required for data volumes for LXC")
+ cls.unsupportedStorageType = True
+ return
cls.disk_offering = DiskOffering.create(
cls.api_client,
cls.services["disk_offering"]
)
+ cls._cleanup.append(cls.disk_offering)
template = get_template(
cls.api_client,
cls.zone.id,
@@ -420,11 +428,13 @@ class TestAttachDetachVolume(cloudstackTestCase):
cls.services["account"],
domainid=cls.domain.id
)
+ cls._cleanup.append(cls.account)
cls.service_offering = ServiceOffering.create(
cls.api_client,
cls.services["service_offering"]
)
+ cls._cleanup.append(cls.service_offering)
cls.virtual_machine = VirtualMachine.create(
cls.api_client,
cls.services["virtual_machine"],
@@ -432,11 +442,6 @@ class TestAttachDetachVolume(cloudstackTestCase):
domainid=cls.account.domainid,
serviceofferingid=cls.service_offering.id,
)
- cls._cleanup = [
- cls.service_offering,
- cls.disk_offering,
- cls.account
- ]
def setUp(self):
@@ -444,6 +449,9 @@ class TestAttachDetachVolume(cloudstackTestCase):
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+ if self.unsupportedStorageType:
+ self.skipTest("RBD storage type is required for data volumes for LXC")
+
def tearDown(self):
# Clean up, terminate the created volumes
cleanup_resources(self.apiclient, self.cleanup)
@@ -612,14 +620,18 @@ class TestAttachVolumeISO(cloudstackTestCase):
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.pod = get_pod(cls.api_client, cls.zone.id)
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.unsupportedStorageType = False
cls.hypervisor = cls.testClient.getHypervisorInfo()
if cls.hypervisor.lower() == 'lxc':
if not find_storage_pool_type(cls.api_client, storagetype='rbd'):
- raise unittest.SkipTest("RBD storage type is required for data volumes for LXC")
+ cls.unsupportedStorageType = True
+ return
cls.disk_offering = DiskOffering.create(
cls.api_client,
cls.services["disk_offering"]
)
+ cls._cleanup.append(cls.disk_offering)
template = get_template(
cls.api_client,
cls.zone.id,
@@ -654,11 +666,13 @@ class TestAttachVolumeISO(cloudstackTestCase):
cls.services["account"],
domainid=cls.domain.id
)
+ cls._cleanup.append(cls.account)
cls.service_offering = ServiceOffering.create(
cls.api_client,
cls.services["service_offering"]
)
+ cls._cleanup.append(cls.service_offering)
cls.virtual_machine = VirtualMachine.create(
cls.api_client,
cls.services["virtual_machine"],
@@ -666,11 +680,6 @@ class TestAttachVolumeISO(cloudstackTestCase):
domainid=cls.account.domainid,
serviceofferingid=cls.service_offering.id,
)
- cls._cleanup = [
- cls.service_offering,
- cls.disk_offering,
- cls.account
- ]
@classmethod
def tearDownClass(cls):
@@ -685,6 +694,9 @@ class TestAttachVolumeISO(cloudstackTestCase):
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+ if self.unsupportedStorageType:
+ self.skipTest("RBD storage type is required for data volumes for LXC")
+
def tearDown(self):
try:
# Clean up, terminate the created instance, volumes and snapshots
@@ -830,14 +842,18 @@ class TestVolumes(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.unsupportedStorageType = False
cls.hypervisor = cls.testClient.getHypervisorInfo()
if cls.hypervisor.lower() == 'lxc':
if not find_storage_pool_type(cls.api_client, storagetype='rbd'):
- raise unittest.SkipTest("RBD storage type is required for data volumes for LXC")
+ cls.unsupportedStorageType = True
+ return
cls.disk_offering = DiskOffering.create(
cls.api_client,
cls.services["disk_offering"]
)
+ cls._cleanup.append(cls.disk_offering)
template = get_template(
cls.api_client,
cls.zone.id,
@@ -855,11 +871,13 @@ class TestVolumes(cloudstackTestCase):
cls.services["account"],
domainid=cls.domain.id
)
+ cls._cleanup.append(cls.account)
cls.service_offering = ServiceOffering.create(
cls.api_client,
cls.services["service_offering"]
)
+ cls._cleanup.append(cls.service_offering)
cls.virtual_machine = VirtualMachine.create(
cls.api_client,
cls.services["virtual_machine"],
@@ -876,11 +894,6 @@ class TestVolumes(cloudstackTestCase):
domainid=cls.account.domainid,
diskofferingid=cls.disk_offering.id
)
- cls._cleanup = [
- cls.service_offering,
- cls.disk_offering,
- cls.account
- ]
@classmethod
def tearDownClass(cls):
@@ -894,6 +907,9 @@ class TestVolumes(cloudstackTestCase):
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+ if self.unsupportedStorageType:
+ self.skipTest("RBD storage type is required for data volumes for LXC")
+
def tearDown(self):
# Clean up, terminate the created volumes
cleanup_resources(self.apiclient, self.cleanup)
@@ -1147,15 +1163,19 @@ class TestDeployVmWithCustomDisk(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.unsupportedStorageType = False
cls.hypervisor = cls.testClient.getHypervisorInfo()
if cls.hypervisor.lower() == 'lxc':
if not find_storage_pool_type(cls.api_client, storagetype='rbd'):
- raise unittest.SkipTest("RBD storage type is required for data volumes for LXC")
+ cls.unsupportedStorageType = True
+ return
cls.disk_offering = DiskOffering.create(
cls.api_client,
cls.services["disk_offering"],
custom=True
)
+ cls._cleanup.append(cls.disk_offering)
template = get_template(
cls.api_client,
cls.zone.id,
@@ -1171,16 +1191,13 @@ class TestDeployVmWithCustomDisk(cloudstackTestCase):
cls.services["account"],
domainid=cls.domain.id
)
+ cls._cleanup.append(cls.account)
cls.service_offering = ServiceOffering.create(
cls.api_client,
cls.services["service_offering"]
)
- cls._cleanup = [
- cls.service_offering,
- cls.disk_offering,
- cls.account
- ]
+ cls._cleanup.append(cls.service_offering)
def setUp(self):
@@ -1188,6 +1205,9 @@ class TestDeployVmWithCustomDisk(cloudstackTestCase):
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+ if self.unsupportedStorageType:
+ self.skipTest("RBD storage type is required for data volumes for LXC")
+
@attr(tags=["advanced", "configuration", "advancedns", "simulator",
"api", "basic", "eip", "sg"])
def test_deployVmWithCustomDisk(self):
@@ -1285,10 +1305,13 @@ class TestMigrateVolume(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.unsupportedStorageType = False
cls.hypervisor = cls.testClient.getHypervisorInfo()
if cls.hypervisor.lower() == 'lxc':
if not find_storage_pool_type(cls.api_client, storagetype='rbd'):
- raise unittest.SkipTest("RBD storage type is required for data volumes for LXC")
+ cls.unsupportedStorageType = True
+ return
cls.disk_offering = DiskOffering.create(
cls.api_client,
cls.services["disk_offering"]
@@ -1340,6 +1363,9 @@ class TestMigrateVolume(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.unsupportedStorageType:
+ self.skipTest("RBD storage type is required for data volumes for LXC")
return
def tearDown(self):
[4/4] git commit: updated refs/heads/master to 6c3dc15
Posted by ga...@apache.org.
CLOUDSTACK-8394: Skip test cases through setUp() instead of setUpClass()
Signed-off-by: Gaurav Aradhye <ga...@clogeny.com>
This closes #231
Project: http://git-wip-us.apache.org/repos/asf/cloudstack/repo
Commit: http://git-wip-us.apache.org/repos/asf/cloudstack/commit/6c3dc15f
Tree: http://git-wip-us.apache.org/repos/asf/cloudstack/tree/6c3dc15f
Diff: http://git-wip-us.apache.org/repos/asf/cloudstack/diff/6c3dc15f
Branch: refs/heads/master
Commit: 6c3dc15fcbf966d3ff626cf190d5ad8bbd06faf1
Parents: 464af9c
Author: Gaurav Aradhye <ga...@clogeny.com>
Authored: Wed May 6 16:02:31 2015 +0530
Committer: Gaurav Aradhye <ga...@clogeny.com>
Committed: Wed May 6 16:25:32 2015 +0530
----------------------------------------------------------------------
test/integration/component/test_projects.py | 1831 ++++++++++--------
.../component/test_snapshot_limits.py | 1 -
test/integration/component/test_templates.py | 562 +++---
test/integration/component/test_usage.py | 29 +-
test/integration/component/test_volumes.py | 90 +-
test/integration/component/test_vpc_routers.py | 1679 ++++++++--------
6 files changed, 2204 insertions(+), 1988 deletions(-)
----------------------------------------------------------------------
[3/4] git commit: updated refs/heads/master to 6c3dc15
Posted by ga...@apache.org.
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/6c3dc15f/test/integration/component/test_projects.py
----------------------------------------------------------------------
diff --git a/test/integration/component/test_projects.py b/test/integration/component/test_projects.py
index cacb5a4..e38bccc 100644
--- a/test/integration/component/test_projects.py
+++ b/test/integration/component/test_projects.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
@@ -16,94 +16,104 @@
# under the License.
""" P1 tests for Project
"""
-#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.sshClient import SshClient
-import datetime
+from marvin.cloudstackTestCase import cloudstackTestCase
+from marvin.lib.utils import cleanup_resources
+from marvin.lib.base import (Account,
+ ServiceOffering,
+ Project,
+ VirtualMachine,
+ Volume,
+ DiskOffering,
+ Configurations,
+ Domain)
+from marvin.lib.common import (get_domain,
+ get_zone,
+ get_template,
+ list_configurations)
+import time
class Services:
+
"""Test Project Services
"""
def __init__(self):
self.services = {
- "domain": {
- "name": "Domain",
- },
- "project": {
- "name": "Project",
- "displaytext": "Test project",
- },
- "mgmt_server": {
- "ipaddress": '192.168.100.21',
- "username": 'root',
- "password": 'password',
- "port": 22,
- },
- "account": {
- "email": "administrator@clogeny.com",
- "firstname": "Test",
- "lastname": "User",
- "username": "test",
- # Random characters are appended for unique
- # username
- "password": "password",
- },
- "user": {
- "email": "administrator@clogeny.com",
- "firstname": "User",
- "lastname": "User",
- "username": "User",
- # Random characters are appended for unique
- # username
- "password": "password",
- },
- "disk_offering": {
- "displaytext": "Tiny Disk Offering",
- "name": "Tiny Disk Offering",
- "disksize": 1
- },
- "volume": {
- "diskname": "Test Volume",
- },
- "service_offering": {
- "name": "Tiny Instance",
- "displaytext": "Tiny Instance",
- "cpunumber": 1,
- "cpuspeed": 100, # in MHz
+ "domain": {
+ "name": "Domain",
+ },
+ "project": {
+ "name": "Project",
+ "displaytext": "Test project",
+ },
+ "mgmt_server": {
+ "ipaddress": '192.168.100.21',
+ "username": 'root',
+ "password": 'password',
+ "port": 22,
+ },
+ "account": {
+ "email": "administrator@clogeny.com",
+ "firstname": "Test",
+ "lastname": "User",
+ "username": "test",
+ # Random characters are appended for unique
+ # username
+ "password": "password",
+ },
+ "user": {
+ "email": "administrator@clogeny.com",
+ "firstname": "User",
+ "lastname": "User",
+ "username": "User",
+ # Random characters are appended for unique
+ # username
+ "password": "password",
+ },
+ "disk_offering": {
+ "displaytext": "Tiny Disk Offering",
+ "name": "Tiny Disk Offering",
+ "disksize": 1
+ },
+ "volume": {
+ "diskname": "Test Volume",
+ },
+ "service_offering": {
+ "name": "Tiny Instance",
+ "displaytext": "Tiny Instance",
+ "cpunumber": 1,
+ "cpuspeed": 100, # in MHz
"memory": 128, # In MBs
- },
- "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,
- }
+ },
+ "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 TestMultipleProjectCreation(cloudstackTestCase):
@classmethod
def setUpClass(cls):
- cls.testClient = super(TestMultipleProjectCreation, cls).getClsTestClient()
+ cls.testClient = super(
+ TestMultipleProjectCreation,
+ cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
cls.services = Services().services
@@ -111,38 +121,38 @@ class TestMultipleProjectCreation(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
-
+ cls._cleanup = []
+ cls.isGlobalSettingInvalid = False
configs = Configurations.list(
- cls.api_client,
- name='project.invite.required'
- )
+ cls.api_client,
+ name='project.invite.required'
+ )
- if not isinstance(configs, list):
- raise unittest.SkipTest("List configurations has no config: project.invite.required")
- elif (configs[0].value).lower() != 'false':
- raise unittest.SkipTest("'project.invite.required' should be set to false")
+ if (configs[0].value).lower() != 'false':
+ cls.isGlobalSettingInvalid = True
+ return
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
+ )
+ cls._cleanup.append(cls.account)
cls.user = Account.create(
- cls.api_client,
- cls.services["account"],
- admin=True,
- domainid=cls.domain.id
- )
-
- cls._cleanup = [cls.account, cls.user]
+ cls.api_client,
+ cls.services["account"],
+ admin=True,
+ domainid=cls.domain.id
+ )
+ cls._cleanup.append(cls.user)
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)
@@ -152,17 +162,28 @@ class TestMultipleProjectCreation(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.isGlobalSettingInvalid:
+ self.skipTest("'project.invite.required' should be set to false")
return
def tearDown(self):
try:
- #Clean up, terminate the created accounts, domains etc
+ # Clean up, terminate the created accounts, domains etc
cleanup_resources(self.apiclient, self.cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
- @attr(tags=["advanced", "basic", "sg", "eip", "advancedns", "simulator"], required_hardware="false")
+ @attr(
+ tags=[
+ "advanced",
+ "basic",
+ "sg",
+ "eip",
+ "advancedns",
+ "simulator"],
+ required_hardware="false")
def test_01_create_multiple_projects_by_account(self):
""" Verify an account can own multiple projects and can belong to multiple projects
"""
@@ -174,139 +195,142 @@ class TestMultipleProjectCreation(cloudstackTestCase):
# Create project as a domain admin
project_1 = Project.create(
- self.apiclient,
- self.services["project"],
- account=self.account.name,
- domainid=self.account.domainid
- )
+ self.apiclient,
+ self.services["project"],
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
# Cleanup created project at end of test
self.cleanup.append(project_1)
self.debug("Created project with domain admin with ID: %s" %
- project_1.id)
+ project_1.id)
list_projects_reponse = Project.list(
- self.apiclient,
- id=project_1.id,
- listall=True
- )
+ self.apiclient,
+ id=project_1.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_projects_reponse, list),
- True,
- "Check for a valid list projects response"
- )
+ isinstance(list_projects_reponse, list),
+ True,
+ "Check for a valid list projects response"
+ )
list_project = list_projects_reponse[0]
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
self.assertEqual(
- project_1.name,
- list_project.name,
- "Check project name from list response"
- )
+ project_1.name,
+ list_project.name,
+ "Check project name from list response"
+ )
# Create another project as a domain admin
project_2 = Project.create(
- self.apiclient,
- self.services["project"],
- account=self.account.name,
- domainid=self.account.domainid
- )
+ self.apiclient,
+ self.services["project"],
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
# Cleanup created project at end of test
self.cleanup.append(project_2)
self.debug("Created project with domain user with ID: %s" %
- project_2.id)
+ project_2.id)
list_projects_reponse = Project.list(
- self.apiclient,
- id=project_2.id,
- listall=True
- )
+ self.apiclient,
+ id=project_2.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_projects_reponse, list),
- True,
- "Check for a valid list projects response"
- )
+ isinstance(list_projects_reponse, list),
+ True,
+ "Check for a valid list projects response"
+ )
list_project = list_projects_reponse[0]
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
self.assert_(isinstance(self.user.user, list))
- self.assert_(len(self.user.user) > 0, msg="Account %s has no users" % self.user.name)
- self.debug("Adding account %s to project with email %s" % (self.user.name, self.user.user[0].email))
+ self.assert_(len(self.user.user) > 0,
+ msg="Account %s has no users" % self.user.name)
+ self.debug(
+ "Adding account %s to project with email %s" %
+ (self.user.name, self.user.user[0].email))
email = self.user.user[0].email
# Add user to the project
project_1.addAccount(
- self.apiclient,
- self.user.name,
- email
- )
+ self.apiclient,
+ self.user.name,
+ email
+ )
# listProjectAccount to verify the user is added to project or not
accounts_response = Project.listAccounts(
- self.apiclient,
- projectid=project_1.id,
- account=self.user.name,
- )
+ self.apiclient,
+ projectid=project_1.id,
+ account=self.user.name,
+ )
self.debug(accounts_response)
self.assertEqual(
- isinstance(accounts_response, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_response, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_response[0]
self.assertEqual(
- account.role,
- 'Regular',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Regular',
+ "Newly added user is not added as a regular user"
+ )
# Add user to the project
project_2.addAccount(
- self.apiclient,
- self.user.name,
- email
- )
+ self.apiclient,
+ self.user.name,
+ email
+ )
# listProjectAccount to verify the user is added to project or not
accounts_response = Project.listAccounts(
- self.apiclient,
- projectid=project_2.id,
- account=self.user.name,
- )
+ self.apiclient,
+ projectid=project_2.id,
+ account=self.user.name,
+ )
self.debug(accounts_response)
self.assertEqual(
- isinstance(accounts_response, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_response, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_response[0]
self.assertEqual(
- account.role,
- 'Regular',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Regular',
+ "Newly added user is not added as a regular user"
+ )
return
@@ -314,7 +338,9 @@ class TestCrossDomainAccountAdd(cloudstackTestCase):
@classmethod
def setUpClass(cls):
- cls.testClient = super(TestCrossDomainAccountAdd, cls).getClsTestClient()
+ cls.testClient = super(
+ TestCrossDomainAccountAdd,
+ cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
cls.services = Services().services
@@ -323,43 +349,44 @@ class TestCrossDomainAccountAdd(cloudstackTestCase):
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.isGlobalSettingInvalid = False
configs = Configurations.list(
- cls.api_client,
- name='project.invite.required'
- )
+ cls.api_client,
+ name='project.invite.required'
+ )
- if not isinstance(configs, list):
- raise unittest.SkipTest("List configurations has no config: project.invite.required")
- elif (configs[0].value).lower() != 'false':
- raise unittest.SkipTest("'project.invite.required' should be set to false")
+ if (configs[0].value).lower() != 'false':
+ cls.isGlobalSettingInvalid = True
+ return
# Create domains, account etc.
cls.new_domain = Domain.create(
- cls.api_client,
- cls.services["domain"]
- )
+ cls.api_client,
+ cls.services["domain"]
+ )
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
+ )
+ cls._cleanup.append(cls.account)
cls.user = Account.create(
- cls.api_client,
- cls.services["account"],
- admin=True,
- domainid=cls.new_domain.id
- )
-
- cls._cleanup = [cls.account, cls.user]
+ cls.api_client,
+ cls.services["account"],
+ admin=True,
+ domainid=cls.new_domain.id
+ )
+ cls._cleanup.append(cls.user)
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)
@@ -369,17 +396,28 @@ class TestCrossDomainAccountAdd(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.isGlobalSettingInvalid:
+ self.skipTest("'project.invite.required' should be set to false")
return
def tearDown(self):
try:
- #Clean up, terminate the created accounts, domains etc
+ # Clean up, terminate the created accounts, domains etc
cleanup_resources(self.apiclient, self.cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
- @attr(tags=["advanced", "basic", "sg", "eip", "advancedns", "simulator"], required_hardware="false")
+ @attr(
+ tags=[
+ "advanced",
+ "basic",
+ "sg",
+ "eip",
+ "advancedns",
+ "simulator"],
+ required_hardware="false")
def test_02_cross_domain_account_add(self):
""" Verify No cross domain projects
"""
@@ -390,52 +428,52 @@ class TestCrossDomainAccountAdd(cloudstackTestCase):
# Create project as a domain admin
project = Project.create(
- self.apiclient,
- self.services["project"],
- account=self.account.name,
- domainid=self.account.domainid
- )
+ self.apiclient,
+ self.services["project"],
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
# Cleanup created project at end of test
self.cleanup.append(project)
self.debug("Created project with domain admin with ID: %s" %
- project.id)
+ project.id)
list_projects_reponse = Project.list(
- self.apiclient,
- id=project.id,
- listall=True
- )
+ self.apiclient,
+ id=project.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_projects_reponse, list),
- True,
- "Check for a valid list projects response"
- )
+ isinstance(list_projects_reponse, list),
+ True,
+ "Check for a valid list projects response"
+ )
list_project = list_projects_reponse[0]
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
self.assertEqual(
- project.name,
- list_project.name,
- "Check project name from list response"
- )
+ project.name,
+ list_project.name,
+ "Check project name from list response"
+ )
self.debug("Adding user: %s from domain: %s to project: %s" % (
- self.user.name,
- self.user.domainid,
- project.id
- ))
+ self.user.name,
+ self.user.domainid,
+ project.id
+ ))
with self.assertRaises(Exception):
# Add user to the project from different domain
project.addAccount(
- self.apiclient,
- self.user.name
- )
+ self.apiclient,
+ self.user.name
+ )
self.debug("User add to project failed!")
return
@@ -444,7 +482,9 @@ class TestDeleteAccountWithProject(cloudstackTestCase):
@classmethod
def setUpClass(cls):
- cls.testClient = super(TestDeleteAccountWithProject, cls).getClsTestClient()
+ cls.testClient = super(
+ TestDeleteAccountWithProject,
+ cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
cls.services = Services().services
@@ -453,30 +493,31 @@ class TestDeleteAccountWithProject(cloudstackTestCase):
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.isGlobalSettingInvalid = False
configs = Configurations.list(
- cls.api_client,
- name='project.invite.required'
- )
+ cls.api_client,
+ name='project.invite.required'
+ )
- if not isinstance(configs, list):
- raise unittest.SkipTest("List configurations has no config: project.invite.required")
- elif (configs[0].value).lower() != 'false':
- raise unittest.SkipTest("'project.invite.required' should be set to false")
+ if (configs[0].value).lower() != 'false':
+ cls.isGlobalSettingInvalid = True
+ return
# Create account
cls.account = Account.create(
- cls.api_client,
- cls.services["account"],
- admin=True,
- domainid=cls.domain.id
- )
- cls._cleanup = [cls.account]
+ cls.api_client,
+ cls.services["account"],
+ admin=True,
+ domainid=cls.domain.id
+ )
+ cls._cleanup.append(cls.account)
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)
@@ -486,17 +527,28 @@ class TestDeleteAccountWithProject(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.isGlobalSettingInvalid:
+ self.skipTest("'project.invite.required' should be set to false")
return
def tearDown(self):
try:
- #Clean up, terminate the created accounts, domains etc
+ # Clean up, terminate the created accounts, domains etc
cleanup_resources(self.apiclient, self.cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
- @attr(tags=["advanced", "basic", "sg", "eip", "advancedns", "simulator"], required_hardware="false")
+ @attr(
+ tags=[
+ "advanced",
+ "basic",
+ "sg",
+ "eip",
+ "advancedns",
+ "simulator"],
+ required_hardware="false")
def test_03_delete_account_with_project(self):
""" Test As long as the project exists, its owner can't be removed
"""
@@ -507,45 +559,45 @@ class TestDeleteAccountWithProject(cloudstackTestCase):
# Create project as a domain admin
project = Project.create(
- self.apiclient,
- self.services["project"],
- account=self.account.name,
- domainid=self.account.domainid
- )
+ self.apiclient,
+ self.services["project"],
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
# Cleanup created project at end of test
self.cleanup.append(project)
self.debug("Created project with domain admin with ID: %s" %
- project.id)
+ project.id)
list_projects_reponse = Project.list(
- self.apiclient,
- id=project.id,
- listall=True
- )
+ self.apiclient,
+ id=project.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_projects_reponse, list),
- True,
- "Check for a valid list projects response"
- )
+ isinstance(list_projects_reponse, list),
+ True,
+ "Check for a valid list projects response"
+ )
list_project = list_projects_reponse[0]
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
self.assertEqual(
- project.name,
- list_project.name,
- "Check project name from list response"
- )
+ project.name,
+ list_project.name,
+ "Check project name from list response"
+ )
# Deleting account who is owner of the project
with self.assertRaises(Exception):
self.account.delete(self.apiclient)
self.debug("Deleting account %s failed!" %
- self.account.name)
+ self.account.name)
return
@@ -553,7 +605,9 @@ class TestDeleteDomainWithProject(cloudstackTestCase):
@classmethod
def setUpClass(cls):
- cls.testClient = super(TestDeleteDomainWithProject, cls).getClsTestClient()
+ cls.testClient = super(
+ TestDeleteDomainWithProject,
+ cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
cls.services = Services().services
@@ -561,35 +615,35 @@ class TestDeleteDomainWithProject(cloudstackTestCase):
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.isGlobalSettingInvalid = False
configs = Configurations.list(
- cls.api_client,
- name='project.invite.required'
- )
+ cls.api_client,
+ name='project.invite.required'
+ )
- if not isinstance(configs, list):
- raise unittest.SkipTest("List configurations has no config: project.invite.required")
- elif (configs[0].value).lower() != 'false':
- raise unittest.SkipTest("'project.invite.required' should be set to false")
+ if (configs[0].value).lower() != 'false':
+ cls.isGlobalSettingInvalid = True
+ return
# Create account
cls.domain = Domain.create(
- cls.api_client,
- cls.services["domain"]
- )
+ cls.api_client,
+ cls.services["domain"]
+ )
cls.account = Account.create(
- cls.api_client,
- cls.services["account"],
- admin=True,
- domainid=cls.domain.id
- )
- cls._cleanup = []
+ cls.api_client,
+ cls.services["account"],
+ admin=True,
+ domainid=cls.domain.id
+ )
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)
@@ -599,17 +653,28 @@ class TestDeleteDomainWithProject(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.isGlobalSettingInvalid:
+ self.skipTest("'project.invite.required' should be set to false")
return
def tearDown(self):
try:
- #Clean up, terminate the created accounts, domains etc
+ # Clean up, terminate the created accounts, domains etc
cleanup_resources(self.apiclient, self.cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
- @attr(tags=["advanced", "basic", "sg", "eip", "advancedns", "simulator"], required_hardware="false")
+ @attr(
+ tags=[
+ "advanced",
+ "basic",
+ "sg",
+ "eip",
+ "advancedns",
+ "simulator"],
+ required_hardware="false")
def test_04_delete_domain_with_project(self):
""" Test Verify delete domain with cleanup=true should delete projects
belonging to the domain
@@ -621,39 +686,39 @@ class TestDeleteDomainWithProject(cloudstackTestCase):
# Create project as a domain admin
project = Project.create(
- self.apiclient,
- self.services["project"],
- account=self.account.name,
- domainid=self.account.domainid
- )
+ self.apiclient,
+ self.services["project"],
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
# Cleanup created project at end of test
self.debug("Created project with domain admin with ID: %s" %
- project.id)
+ project.id)
list_projects_reponse = Project.list(
- self.apiclient,
- id=project.id,
- listall=True
- )
+ self.apiclient,
+ id=project.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_projects_reponse, list),
- True,
- "Check for a valid list projects response"
- )
+ isinstance(list_projects_reponse, list),
+ True,
+ "Check for a valid list projects response"
+ )
list_project = list_projects_reponse[0]
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
self.assertEqual(
- project.name,
- list_project.name,
- "Check project name from list response"
- )
+ project.name,
+ list_project.name,
+ "Check project name from list response"
+ )
self.debug("Deleting domain: %s forcefully" % self.domain.name)
# Delete domain with cleanup=True
@@ -661,31 +726,31 @@ class TestDeleteDomainWithProject(cloudstackTestCase):
self.debug("Removed domain: %s" % self.domain.name)
interval = list_configurations(
- self.apiclient,
- name='account.cleanup.interval'
- )
+ self.apiclient,
+ name='account.cleanup.interval'
+ )
self.assertEqual(
- isinstance(interval, list),
- True,
- "Check if account.cleanup.interval config present"
- )
+ isinstance(interval, list),
+ True,
+ "Check if account.cleanup.interval config present"
+ )
self.debug(
- "Sleep for account cleanup interval: %s" %
- interval[0].value)
+ "Sleep for account cleanup interval: %s" %
+ interval[0].value)
# Sleep to ensure that all resources are deleted
time.sleep(int(interval[0].value))
# Project should be deleted as part of domain cleanup
list_projects_reponse = Project.list(
- self.apiclient,
- id=project.id,
- listall=True
- )
+ self.apiclient,
+ id=project.id,
+ listall=True
+ )
self.assertEqual(
- list_projects_reponse,
- None,
- "Project should be deleted as part of domain cleanup"
- )
+ list_projects_reponse,
+ None,
+ "Project should be deleted as part of domain cleanup"
+ )
return
@@ -702,36 +767,38 @@ class TestProjectOwners(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.isGlobalSettingInvalid = False
configs = Configurations.list(
- cls.api_client,
- name='project.invite.required'
- )
+ cls.api_client,
+ name='project.invite.required'
+ )
- if not isinstance(configs, list):
- raise unittest.SkipTest("List configurations has no config: project.invite.required")
- elif (configs[0].value).lower() != 'false':
- raise unittest.SkipTest("'project.invite.required' should be set to false")
+ if (configs[0].value).lower() != 'false':
+ cls.isGlobalSettingInvalid = True
+ return
# Create accounts
cls.admin = 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
+ )
+ cls._cleanup.append(cls.admin)
cls.new_admin = Account.create(
- cls.api_client,
- cls.services["account"],
- admin=True,
- domainid=cls.domain.id
- )
- cls._cleanup = [cls.admin, cls.new_admin]
+ cls.api_client,
+ cls.services["account"],
+ admin=True,
+ domainid=cls.domain.id
+ )
+ cls._cleanup.append(cls.new_admin)
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)
@@ -741,17 +808,28 @@ class TestProjectOwners(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.isGlobalSettingInvalid:
+ self.skipTest("'project.invite.required' should be set to false")
return
def tearDown(self):
try:
- #Clean up, terminate the created accounts, domains etc
+ # Clean up, terminate the created accounts, domains etc
cleanup_resources(self.apiclient, self.cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
- @attr(tags=["advanced", "basic", "sg", "eip", "advancedns", "simulator"], required_hardware="false")
+ @attr(
+ tags=[
+ "advanced",
+ "basic",
+ "sg",
+ "eip",
+ "advancedns",
+ "simulator"],
+ required_hardware="false")
def test_05_user_project_owner_promotion(self):
""" Test Verify a project user can be later promoted to become a
owner
@@ -764,136 +842,144 @@ class TestProjectOwners(cloudstackTestCase):
# Create project as a domain admin
project = Project.create(
- self.apiclient,
- self.services["project"],
- account=self.admin.name,
- domainid=self.admin.domainid
- )
+ self.apiclient,
+ self.services["project"],
+ account=self.admin.name,
+ domainid=self.admin.domainid
+ )
self.cleanup.append(project)
# Cleanup created project at end of test
self.debug("Created project with domain admin with ID: %s" %
- project.id)
+ project.id)
list_projects_reponse = Project.list(
- self.apiclient,
- id=project.id,
- listall=True
- )
+ self.apiclient,
+ id=project.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_projects_reponse, list),
- True,
- "Check for a valid list projects response"
- )
+ isinstance(list_projects_reponse, list),
+ True,
+ "Check for a valid list projects response"
+ )
list_project = list_projects_reponse[0]
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
self.assertEqual(
- project.name,
- list_project.name,
- "Check project name from list response"
- )
+ project.name,
+ list_project.name,
+ "Check project name from list response"
+ )
self.debug("Adding %s user to project: %s" % (
- self.new_admin.name,
- project.name
- ))
+ self.new_admin.name,
+ project.name
+ ))
# Add user to the project
project.addAccount(
- self.apiclient,
- self.new_admin.name,
- )
+ self.apiclient,
+ self.new_admin.name,
+ )
# listProjectAccount to verify the user is added to project or not
accounts_reponse = Project.listAccounts(
- self.apiclient,
- projectid=project.id,
- account=self.new_admin.name,
- )
+ self.apiclient,
+ projectid=project.id,
+ account=self.new_admin.name,
+ )
self.debug(accounts_reponse)
self.assertEqual(
- isinstance(accounts_reponse, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_reponse, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_reponse[0]
self.assertEqual(
- account.role,
- 'Regular',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Regular',
+ "Newly added user is not added as a regular user"
+ )
# Update the project with new admin
project.update(
- self.apiclient,
- account=self.new_admin.name
- )
+ self.apiclient,
+ account=self.new_admin.name
+ )
# listProjectAccount to verify the user is new admin of the project
accounts_reponse = Project.listAccounts(
- self.apiclient,
- projectid=project.id,
- account=self.new_admin.name,
- )
+ self.apiclient,
+ projectid=project.id,
+ account=self.new_admin.name,
+ )
self.debug(accounts_reponse)
self.assertEqual(
- isinstance(accounts_reponse, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_reponse, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_reponse[0]
self.assertEqual(
- account.role,
- 'Admin',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Admin',
+ "Newly added user is not added as a regular user"
+ )
# listProjectAccount to verify old user becomes a regular user
accounts_reponse = Project.listAccounts(
- self.apiclient,
- projectid=project.id,
- account=self.admin.name,
- )
+ self.apiclient,
+ projectid=project.id,
+ account=self.admin.name,
+ )
self.debug(accounts_reponse)
self.assertEqual(
- isinstance(accounts_reponse, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_reponse, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_reponse[0]
self.assertEqual(
- account.role,
- 'Regular',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Regular',
+ "Newly added user is not added as a regular user"
+ )
return
- @attr(tags=["advanced", "basic", "sg", "eip", "advancedns", "simulator"], required_hardware="false")
+ @attr(
+ tags=[
+ "advanced",
+ "basic",
+ "sg",
+ "eip",
+ "advancedns",
+ "simulator"],
+ required_hardware="false")
def test_06_max_one_project_owner(self):
""" Test Verify there can only be one owner of a project at a time
"""
@@ -905,211 +991,211 @@ class TestProjectOwners(cloudstackTestCase):
# Create project as a domain admin
project = Project.create(
- self.apiclient,
- self.services["project"],
- account=self.admin.name,
- domainid=self.admin.domainid
- )
+ self.apiclient,
+ self.services["project"],
+ account=self.admin.name,
+ domainid=self.admin.domainid
+ )
# Cleanup created project at end of test
self.cleanup.append(project)
self.debug("Created project with domain admin with ID: %s" %
- project.id)
+ project.id)
self.user = Account.create(
- self.apiclient,
- self.services["account"],
- admin=True,
- domainid=self.domain.id
- )
+ self.apiclient,
+ self.services["account"],
+ admin=True,
+ domainid=self.domain.id
+ )
self.cleanup.append(self.user)
self.debug("Created account with ID: %s" %
- self.user.name)
+ self.user.name)
list_projects_reponse = Project.list(
- self.apiclient,
- id=project.id,
- listall=True
- )
+ self.apiclient,
+ id=project.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_projects_reponse, list),
- True,
- "Check for a valid list projects response"
- )
+ isinstance(list_projects_reponse, list),
+ True,
+ "Check for a valid list projects response"
+ )
list_project = list_projects_reponse[0]
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
self.assertEqual(
- project.name,
- list_project.name,
- "Check project name from list response"
- )
+ project.name,
+ list_project.name,
+ "Check project name from list response"
+ )
self.debug("Adding %s user to project: %s" % (
- self.new_admin.name,
- project.name
- ))
+ self.new_admin.name,
+ project.name
+ ))
# Add user to the project
project.addAccount(
- self.apiclient,
- self.new_admin.name,
- )
+ self.apiclient,
+ self.new_admin.name,
+ )
# listProjectAccount to verify the user is added to project or not
accounts_reponse = Project.listAccounts(
- self.apiclient,
- projectid=project.id,
- account=self.new_admin.name,
- )
+ self.apiclient,
+ projectid=project.id,
+ account=self.new_admin.name,
+ )
self.debug(accounts_reponse)
self.assertEqual(
- isinstance(accounts_reponse, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_reponse, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_reponse[0]
self.assertEqual(
- account.role,
- 'Regular',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Regular',
+ "Newly added user is not added as a regular user"
+ )
self.debug("Updating project with new Admin: %s" %
- self.new_admin.name)
+ self.new_admin.name)
# Update the project with new admin
project.update(
- self.apiclient,
- account=self.new_admin.name
- )
+ self.apiclient,
+ account=self.new_admin.name
+ )
# listProjectAccount to verify the user is new admin of the project
accounts_reponse = Project.listAccounts(
- self.apiclient,
- projectid=project.id,
- account=self.new_admin.name,
- )
+ self.apiclient,
+ projectid=project.id,
+ account=self.new_admin.name,
+ )
self.assertEqual(
- isinstance(accounts_reponse, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_reponse, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_reponse[0]
self.assertEqual(
- account.role,
- 'Admin',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Admin',
+ "Newly added user is not added as a regular user"
+ )
self.debug("Adding %s user to project: %s" % (
- self.user.name,
- project.name
- ))
+ self.user.name,
+ project.name
+ ))
# Add user to the project
project.addAccount(
- self.apiclient,
- self.user.name,
- )
+ self.apiclient,
+ self.user.name,
+ )
# listProjectAccount to verify the user is added to project or not
accounts_reponse = Project.listAccounts(
- self.apiclient,
- projectid=project.id,
- account=self.user.name,
- )
+ self.apiclient,
+ projectid=project.id,
+ account=self.user.name,
+ )
self.assertEqual(
- isinstance(accounts_reponse, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_reponse, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_reponse[0]
self.assertEqual(
- account.role,
- 'Regular',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Regular',
+ "Newly added user is not added as a regular user"
+ )
self.debug("Updating project with new Admin: %s" %
- self.user.name)
+ self.user.name)
# Update the project with new admin
project.update(
- self.apiclient,
- account=self.user.name
- )
+ self.apiclient,
+ account=self.user.name
+ )
# listProjectAccount to verify the user is new admin of the project
accounts_reponse = Project.listAccounts(
- self.apiclient,
- projectid=project.id,
- account=self.user.name,
- )
+ self.apiclient,
+ projectid=project.id,
+ account=self.user.name,
+ )
self.debug(accounts_reponse)
self.assertEqual(
- isinstance(accounts_reponse, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_reponse, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_reponse[0]
self.assertEqual(
- account.role,
- 'Admin',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Admin',
+ "Newly added user is not added as a regular user"
+ )
# listProjectAccount to verify old user becomes a regular user
accounts_reponse = Project.listAccounts(
- self.apiclient,
- projectid=project.id,
- account=self.new_admin.name,
- )
+ self.apiclient,
+ projectid=project.id,
+ account=self.new_admin.name,
+ )
self.assertEqual(
- isinstance(accounts_reponse, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_reponse, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_reponse[0]
self.assertEqual(
- account.role,
- 'Regular',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Regular',
+ "Newly added user is not added as a regular user"
+ )
return
@@ -1126,41 +1212,43 @@ class TestProjectResources(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
cls.services['mode'] = cls.zone.networktype
+ cls._cleanup = []
+ cls.isGlobalSettingInvalid = False
configs = Configurations.list(
- cls.api_client,
- name='project.invite.required'
- )
+ cls.api_client,
+ name='project.invite.required'
+ )
- if not isinstance(configs, list):
- raise unittest.SkipTest("List configurations has no config: project.invite.required")
- elif (configs[0].value).lower() != 'false':
- raise unittest.SkipTest("'project.invite.required' should be set to false")
+ if (configs[0].value).lower() != 'false':
+ cls.isGlobalSettingInvalid = True
+ return
# Create account, disk offering etc.
cls.disk_offering = DiskOffering.create(
- cls.api_client,
- cls.services["disk_offering"]
- )
+ cls.api_client,
+ cls.services["disk_offering"]
+ )
+ cls._cleanup.append(cls.disk_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
+ )
+ cls._cleanup.append(cls.account)
cls.user = Account.create(
- cls.api_client,
- cls.services["account"],
- admin=True,
- domainid=cls.domain.id
- )
- cls._cleanup = [cls.account, cls.disk_offering]
+ cls.api_client,
+ cls.services["account"],
+ admin=True,
+ domainid=cls.domain.id
+ )
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)
@@ -1170,17 +1258,28 @@ class TestProjectResources(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.isGlobalSettingInvalid:
+ self.skipTest("'project.invite.required' should be set to false")
return
def tearDown(self):
try:
- #Clean up, terminate the created accounts, domains etc
+ # Clean up, terminate the created accounts, domains etc
cleanup_resources(self.apiclient, self.cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
- @attr(tags=["advanced", "basic", "sg", "eip", "advancedns", "simulator"], required_hardware="false")
+ @attr(
+ tags=[
+ "advanced",
+ "basic",
+ "sg",
+ "eip",
+ "advancedns",
+ "simulator"],
+ required_hardware="false")
def test_07_project_resources_account_delete(self):
""" Test Verify after an account is removed from the project, all its resources stay with the project.
"""
@@ -1192,82 +1291,82 @@ class TestProjectResources(cloudstackTestCase):
# Create project as a domain admin
project = Project.create(
- self.apiclient,
- self.services["project"],
- account=self.account.name,
- domainid=self.account.domainid
- )
+ self.apiclient,
+ self.services["project"],
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
# Cleanup created project at end of test
self.cleanup.append(project)
self.debug("Created project with domain admin with ID: %s" %
- project.id)
+ project.id)
list_projects_reponse = Project.list(
- self.apiclient,
- id=project.id,
- listall=True
- )
+ self.apiclient,
+ id=project.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_projects_reponse, list),
- True,
- "Check for a valid list projects response"
- )
+ isinstance(list_projects_reponse, list),
+ True,
+ "Check for a valid list projects response"
+ )
list_project = list_projects_reponse[0]
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
self.assertEqual(
- project.name,
- list_project.name,
- "Check project name from list response"
- )
+ project.name,
+ list_project.name,
+ "Check project name from list response"
+ )
self.debug("Adding %s user to project: %s" % (
- self.user.name,
- project.name
- ))
+ self.user.name,
+ project.name
+ ))
# Add user to the project
project.addAccount(
- self.apiclient,
- self.user.name,
- )
+ self.apiclient,
+ self.user.name,
+ )
# listProjectAccount to verify the user is added to project or not
accounts_response = Project.listAccounts(
- self.apiclient,
- projectid=project.id,
- account=self.user.name,
- )
+ self.apiclient,
+ projectid=project.id,
+ account=self.user.name,
+ )
self.assertEqual(
- isinstance(accounts_response, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_response, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_response[0]
self.assertEqual(
- account.role,
- 'Regular',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Regular',
+ "Newly added user is not added as a regular user"
+ )
# Create some resources(volumes) for the projects
volume = Volume.create(
- self.apiclient,
- self.services["volume"],
- zoneid=self.zone.id,
- diskofferingid=self.disk_offering.id,
- projectid=project.id
- )
+ self.apiclient,
+ self.services["volume"],
+ zoneid=self.zone.id,
+ diskofferingid=self.disk_offering.id,
+ projectid=project.id
+ )
# Delete the project user ie the account
self.user.delete(self.apiclient)
@@ -1275,26 +1374,34 @@ class TestProjectResources(cloudstackTestCase):
volumes = Volume.list(self.apiclient, id=volume.id)
self.assertEqual(
- isinstance(volumes, list),
- True,
- "Check for a valid list volumes response"
- )
+ isinstance(volumes, list),
+ True,
+ "Check for a valid list volumes response"
+ )
self.assertNotEqual(
- len(volumes),
- 0,
- "Check list volumes API response returns a valid list"
- )
+ len(volumes),
+ 0,
+ "Check list volumes API response returns a valid list"
+ )
volume_response = volumes[0]
self.assertEqual(
- volume_response.name,
- volume.name,
- "Volume should exist after project user deletion."
- )
+ volume_response.name,
+ volume.name,
+ "Volume should exist after project user deletion."
+ )
return
- @attr(tags=["advanced", "basic", "sg", "eip", "advancedns", "simulator"], required_hardware="false")
+ @attr(
+ tags=[
+ "advanced",
+ "basic",
+ "sg",
+ "eip",
+ "advancedns",
+ "simulator"],
+ required_hardware="false")
def test_08_cleanup_after_project_delete(self):
""" Test accounts are unassigned from project after project deletion
"""
@@ -1307,92 +1414,92 @@ class TestProjectResources(cloudstackTestCase):
# Create project as a domain admin
project = Project.create(
- self.apiclient,
- self.services["project"],
- account=self.account.name,
- domainid=self.account.domainid
- )
+ self.apiclient,
+ self.services["project"],
+ account=self.account.name,
+ domainid=self.account.domainid
+ )
# Cleanup created project at end of test
self.debug("Created project with domain admin with ID: %s" %
- project.id)
+ project.id)
list_projects_reponse = Project.list(
- self.apiclient,
- id=project.id,
- listall=True
- )
+ self.apiclient,
+ id=project.id,
+ listall=True
+ )
self.assertEqual(
- isinstance(list_projects_reponse, list),
- True,
- "Check for a valid list projects response"
- )
+ isinstance(list_projects_reponse, list),
+ True,
+ "Check for a valid list projects response"
+ )
list_project = list_projects_reponse[0]
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
self.assertEqual(
- project.name,
- list_project.name,
- "Check project name from list response"
- )
+ project.name,
+ list_project.name,
+ "Check project name from list response"
+ )
self.user = Account.create(
- self.apiclient,
- self.services["account"],
- admin=True,
- domainid=self.domain.id
- )
+ self.apiclient,
+ self.services["account"],
+ admin=True,
+ domainid=self.domain.id
+ )
self.cleanup.append(self.user)
self.debug("Adding %s user to project: %s" % (
- self.user.name,
- project.name
- ))
+ self.user.name,
+ project.name
+ ))
# Add user to the project
project.addAccount(
- self.apiclient,
- self.user.name
- )
+ self.apiclient,
+ self.user.name
+ )
# listProjectAccount to verify the user is added to project or not
accounts_reponse = Project.listAccounts(
- self.apiclient,
- projectid=project.id,
- account=self.user.name,
- )
+ self.apiclient,
+ projectid=project.id,
+ account=self.user.name,
+ )
self.assertEqual(
- isinstance(accounts_reponse, list),
- True,
- "Check for a valid list accounts response"
- )
+ isinstance(accounts_reponse, list),
+ True,
+ "Check for a valid list accounts response"
+ )
self.assertNotEqual(
- len(list_projects_reponse),
- 0,
- "Check list project response returns a valid project"
- )
+ len(list_projects_reponse),
+ 0,
+ "Check list project response returns a valid project"
+ )
account = accounts_reponse[0]
self.assertEqual(
- account.role,
- 'Regular',
- "Newly added user is not added as a regular user"
- )
+ account.role,
+ 'Regular',
+ "Newly added user is not added as a regular user"
+ )
# Create some resources(volumes) for the projects
volume = Volume.create(
- self.apiclient,
- self.services["volume"],
- zoneid=self.zone.id,
- diskofferingid=self.disk_offering.id,
- projectid=project.id
- )
+ self.apiclient,
+ self.services["volume"],
+ zoneid=self.zone.id,
+ diskofferingid=self.disk_offering.id,
+ projectid=project.id
+ )
self.debug("Created a volume: %s for project: %s" % (
- volume.id,
- project.name
- ))
+ volume.id,
+ project.name
+ ))
# Delete the project user
self.debug("Deleting project: %s" % project.name)
project.delete(self.apiclient)
@@ -1401,10 +1508,10 @@ class TestProjectResources(cloudstackTestCase):
volumes = Volume.list(self.apiclient, id=volume.id)
self.assertEqual(
- volumes,
- None,
- "Resources (volume) should be deleted as part of cleanup"
- )
+ volumes,
+ None,
+ "Resources (volume) should be deleted as part of cleanup"
+ )
# Accounts should be un-assigned from project,
# so this call will raise an exception: Unable to find the project id=
@@ -1417,7 +1524,9 @@ class TestProjectSuspendActivate(cloudstackTestCase):
@classmethod
def setUpClass(cls):
- cls.testClient = super(TestProjectSuspendActivate, cls).getClsTestClient()
+ cls.testClient = super(
+ TestProjectSuspendActivate,
+ cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
cls.services = Services().services
@@ -1426,64 +1535,63 @@ class TestProjectSuspendActivate(cloudstackTestCase):
cls.domain = get_domain(cls.api_client)
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._cleanup = []
+ cls.isGlobalSettingInvalid = False
configs = Configurations.list(
- cls.api_client,
- name='project.invite.required'
- )
+ cls.api_client,
+ name='project.invite.required'
+ )
- if not isinstance(configs, list):
- raise unittest.SkipTest("List configurations has no config: project.invite.required")
- elif (configs[0].value).lower() != 'false':
- raise unittest.SkipTest("'project.invite.required' should be set to false")
+ if (configs[0].value).lower() != 'false':
+ cls.isGlobalSettingInvalid = True
+ return
# Create account, service offering, disk offering etc.
cls.disk_offering = DiskOffering.create(
- cls.api_client,
- cls.services["disk_offering"]
- )
+ cls.api_client,
+ cls.services["disk_offering"]
+ )
+ cls._cleanup.append(cls.disk_offering)
cls.service_offering = ServiceOffering.create(
- cls.api_client,
- cls.services["service_offering"],
- domainid=cls.domain.id
- )
+ cls.api_client,
+ cls.services["service_offering"],
+ domainid=cls.domain.id
+ )
+ cls._cleanup.append(cls.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
+ )
cls.user = 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
+ )
# Create project as a domain admin
cls.project = Project.create(
- cls.api_client,
- cls.services["project"],
- account=cls.account.name,
- domainid=cls.account.domainid
- )
+ cls.api_client,
+ cls.services["project"],
+ account=cls.account.name,
+ domainid=cls.account.domainid
+ )
+ cls._cleanup.append(cls.project)
+ cls._cleanup.append(cls.account)
+ cls._cleanup.append(cls.user)
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
- cls._cleanup = [
- cls.project,
- cls.account,
- cls.user,
- cls.disk_offering,
- 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)
@@ -1493,17 +1601,28 @@ class TestProjectSuspendActivate(cloudstackTestCase):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.cleanup = []
+
+ if self.isGlobalSettingInvalid:
+ self.skipTest("'project.invite.required' should be set to false")
return
def tearDown(self):
try:
- #Clean up, terminate the created accounts, domains etc
+ # Clean up, terminate the created accounts, domains etc
cleanup_resources(self.apiclient, self.cleanup)
except Exception as e:
raise Exception("Warning: Exception during cleanup : %s" % e)
return
- @attr(tags=["advanced", "basic", "sg", "eip", "advancedns", "simulator"], required_hardware="false")
+ @attr(
+ tags=[
+ "advanced",
+ "basic",
+ "sg",
+ "eip",
+ "advancedns",
+ "simulator"],
+ required_hardware="false")
def test_09_project_suspend(self):
""" Test Verify after an account is removed from the project, all his
resources stay with the project.
@@ -1515,96 +1634,96 @@ class TestProjectSuspendActivate(cloudstackTestCase):
# account deletion.
self.debug("Adding %s user to project: %s" % (
- self.user.name,
- self.project.name
- ))
+ self.user.name,
+ self.project.name
+ ))
# Add user to
<TRUNCATED>