diff --git a/test/integration/component/test_projects.py b/test/integration/component/test_projects.py index cacb5a44de5..e38bccc45ca 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 the project self.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=self.project.id, - account=self.user.name, - ) + self.apiclient, + projectid=self.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(accounts_reponse), - 0, - "Check list project response returns a valid project" - ) + len(accounts_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" + ) virtual_machine = VirtualMachine.create( - self.apiclient, - self.services["virtual_machine"], - templateid=self.template.id, - serviceofferingid=self.service_offering.id, - projectid=self.project.id - ) + self.apiclient, + self.services["virtual_machine"], + templateid=self.template.id, + serviceofferingid=self.service_offering.id, + projectid=self.project.id + ) self.debug("Created a VM: %s for project: %s" % ( - virtual_machine.id, - self.project.id - )) + virtual_machine.id, + self.project.id + )) self.debug("Suspending a project: %s" % self.project.name) self.project.suspend(self.apiclient) # Check status of all VMs associated with project vms = VirtualMachine.list( - self.apiclient, - projectid=self.project.id, - listall=True - ) + self.apiclient, + projectid=self.project.id, + listall=True + ) self.assertEqual( - isinstance(vms, list), - True, - "Check for a valid list accounts response" - ) + isinstance(vms, list), + True, + "Check for a valid list accounts response" + ) self.assertNotEqual( - len(vms), - 0, - "Check list project response returns a valid project" - ) + len(vms), + 0, + "Check list project response returns a valid project" + ) for vm in vms: self.debug("VM ID: %s state: %s" % (vm.id, vm.state)) self.assertEqual( - vm.state, - 'Stopped', - "VM should be in stopped state after project suspension" - ) + vm.state, + 'Stopped', + "VM should be in stopped state after project suspension" + ) self.debug("Attempting to create volume in suspended project") with self.assertRaises(Exception): # 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=self.project.id - ) + Volume.create( + self.apiclient, + self.services["volume"], + zoneid=self.zone.id, + diskofferingid=self.disk_offering.id, + projectid=self.project.id + ) self.debug("Volume creation failed") # Start the stopped VM self.debug("Attempting to start VM: %s in suspended project" % - virtual_machine.id) + virtual_machine.id) with self.assertRaises(Exception): virtual_machine.start(self.apiclient) self.debug("VM start failed!") @@ -1615,32 +1734,40 @@ class TestProjectSuspendActivate(cloudstackTestCase): # Check status of all VMs associated with project vms = VirtualMachine.list( - self.apiclient, - projectid=self.project.id, - listall=True - ) + self.apiclient, + projectid=self.project.id, + listall=True + ) self.assertEqual( - isinstance(vms, list), - True, - "Check for a valid list accounts response" - ) + isinstance(vms, list), + True, + "Check for a valid list accounts response" + ) self.assertNotEqual( - len(vms), - 0, - "Check list project response returns a valid project" - ) + len(vms), + 0, + "Check list project response returns a valid project" + ) for vm in vms: self.debug("VM ID: %s state: %s" % (vm.id, vm.state)) self.assertEqual( - vm.state, - 'Destroyed', - "VM should be in stopped state after project suspension" - ) + vm.state, + 'Destroyed', + "VM should be in stopped state after project suspension" + ) 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_10_project_activation(self): """ Test project activation after suspension """ @@ -1653,41 +1780,41 @@ class TestProjectSuspendActivate(cloudstackTestCase): self.project.activate(self.apiclient) virtual_machine = VirtualMachine.create( - self.apiclient, - self.services["virtual_machine"], - templateid=self.template.id, - serviceofferingid=self.service_offering.id, - projectid=self.project.id - ) + self.apiclient, + self.services["virtual_machine"], + templateid=self.template.id, + serviceofferingid=self.service_offering.id, + projectid=self.project.id + ) self.cleanup.append(virtual_machine) self.debug("Created a VM: %s for project: %s" % ( - virtual_machine.id, - self.project.id - )) + virtual_machine.id, + self.project.id + )) # Check status of all VMs associated with project vms = VirtualMachine.list( - self.apiclient, - id=virtual_machine.id, - listall=True - ) + self.apiclient, + id=virtual_machine.id, + listall=True + ) self.assertEqual( - isinstance(vms, list), - True, - "Check for a valid list accounts response" - ) + isinstance(vms, list), + True, + "Check for a valid list accounts response" + ) self.assertNotEqual( - len(vms), - 0, - "Check list project response returns a valid project" - ) + len(vms), + 0, + "Check list project response returns a valid project" + ) for vm in vms: self.debug("VM ID: %s state: %s" % (vm.id, vm.state)) self.assertEqual( - vm.state, - 'Running', - "VM should be in Running state after project activation" - ) + vm.state, + 'Running', + "VM should be in Running state after project activation" + ) return diff --git a/test/integration/component/test_snapshot_limits.py b/test/integration/component/test_snapshot_limits.py index 6d8c980ffe4..21e15da236a 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, diff --git a/test/integration/component/test_templates.py b/test/integration/component/test_templates.py index eafc1bbb05d..7973766eb03 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.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.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 diff --git a/test/integration/component/test_usage.py b/test/integration/component/test_usage.py index c0d40316331..5eb4c742fca 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): diff --git a/test/integration/component/test_volumes.py b/test/integration/component/test_volumes.py index 1b01e777227..dd8b5e83601 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): diff --git a/test/integration/component/test_vpc_routers.py b/test/integration/component/test_vpc_routers.py index 54f6a83ecde..c27e4b2e3f3 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', - }, + "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' + } - "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)] - ) + 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 - ) + 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 - ) + 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 - ) + 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' - ) + 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_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 - ) + 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' - ) + 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' + ) - nwacl_internet_1 = 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 - router_response = list_routers( - self.api_client, - id=router.id - ) - self.assertEqual( - 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)) - return + self.debug("Waiting for Router mgiration ....") + time.sleep(240) + + # List routers to check state of router + router_response = list_routers( + self.api_client, + id=router.id + ) + self.assertEqual( + 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)) + 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, - "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