From 3609f604473a5b61aec016786ba2dd671e2bd56e Mon Sep 17 00:00:00 2001 From: Doug Clark Date: Thu, 4 Sep 2014 13:28:11 +0000 Subject: [PATCH] CLOUDSTACK-7355 - test_netscaler_configs.py uses hardcoded NetScaler credentials Refactored code to use test_data instead of hardcoded NetScaler credentials Refactored code to remove large scale duplication Fixed some minor logic error in the existing tests This patch has not added or removed any of the original test-cases. Signed-off-by: SrikanteswaraRao Talluri --- .../component/test_netscaler_configs.py | 1435 +++++------------ tools/marvin/marvin/config/test_data.py | 10 +- 2 files changed, 399 insertions(+), 1046 deletions(-) diff --git a/test/integration/component/test_netscaler_configs.py b/test/integration/component/test_netscaler_configs.py index b94e42d4e97..91fb85b21dd 100644 --- a/test/integration/component/test_netscaler_configs.py +++ b/test/integration/component/test_netscaler_configs.py @@ -41,160 +41,25 @@ from marvin.lib.common import (get_domain, from marvin.sshClient import SshClient import time - -class Services: - """Test netscaler 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 - "memory": 128, # In MBs - }, - "virtual_machine": { - "displayname": "TestVM", - "username": "root", - "password": "password", - "ssh_port": 22, - "hypervisor": 'XenServer', - "privateport": 22, - "publicport": 22, - "protocol": 'TCP', - }, - "netscaler": { - "ipaddress": '10.147.60.26', - "username": 'nsroot', - "password": 'nsroot', - "networkdevicetype": 'NetscalerVPXLoadBalancer', - "publicinterface": '0/1', - "privateinterface": '1/1', - "numretries": 2, - "lbdevicededicated": False, - "lbdevicecapacity": 50, - "port": 22, - }, - "netscaler_dedicated": { - "ipaddress": '10.147.60.27', - "username": 'nsroot', - "password": 'nsroot', - "networkdevicetype": 'NetscalerVPXLoadBalancer', - "publicinterface": '0/1', - "privateinterface": '1/1', - "numretries": 2, - "lbdevicededicated": True, - "port": 22, - }, - "network_offering_dedicated": { - "name": 'Netscaler', - "displaytext": 'Netscaler', - "guestiptype": 'Isolated', - "supportedservices": 'Dhcp,Dns,SourceNat,PortForwarding,Vpn,Firewall,Lb,UserData,StaticNat', - "traffictype": 'GUEST', - "availability": 'Optional', - "specifyVlan": False, - "specifyIpRanges": False, - "serviceProviderList" : { - "Dhcp": 'VirtualRouter', - "Dns": 'VirtualRouter', - "SourceNat": 'VirtualRouter', - "PortForwarding": 'VirtualRouter', - "Vpn": 'VirtualRouter', - "Firewall": 'VirtualRouter', - "Lb": 'Netscaler', - "UserData": 'VirtualRouter', - "StaticNat": 'VirtualRouter', - }, - "serviceCapabilityList": { - "SourceNat": { - "SupportedSourceNatTypes": "peraccount" - }, - "lb": { - "SupportedLbIsolation": "dedicated" - }, - }, - }, - "network_offering": { - "name": 'Netscaler', - "displaytext": 'Netscaler', - "guestiptype": 'Isolated', - "supportedservices": 'Dhcp,Dns,SourceNat,PortForwarding,Vpn,Firewall,Lb,UserData,StaticNat', - "traffictype": 'GUEST', - "availability": 'Optional', - "serviceProviderList" : { - "Dhcp": 'VirtualRouter', - "Dns": 'VirtualRouter', - "SourceNat": 'VirtualRouter', - "PortForwarding": 'VirtualRouter', - "Vpn": 'VirtualRouter', - "Firewall": 'VirtualRouter', - "Lb": 'Netscaler', - "UserData": 'VirtualRouter', - "StaticNat": 'VirtualRouter', - }, - }, - "network": { - "name": "Netscaler", - "displaytext": "Netscaler", - }, - "lbrule": { - "name": "SSH", - "alg": "roundrobin", - # Algorithm used for load balancing - "privateport": 22, - "publicport": 22, - "openfirewall": False, - }, - "ostype": 'CentOS 5.3 (64-bit)', - # Cent OS 5.3 (64 bit) - "sleep": 60, - "timeout": 10, - "mode":'advanced' - } - - -class TestAddNetScaler(cloudstackTestCase): - - @classmethod - def setUpClass(cls): - cls.testClient = super(TestAddNetScaler, cls).getClsTestClient() - cls.api_client = cls.testClient.getApiClient() - - 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._cleanup = [] - return - +class _NetScalerBase(cloudstackTestCase): + """Base class for NetScaler tests in this file""" @classmethod def tearDownClass(cls): + """Generic class tear-down method""" try: #Cleanup resources used cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) - return def setUp(self): + """Generic test-case setup method""" self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() self.cleanup = [] - return def tearDown(self): + """Generic test-case tear-down method""" try: self.debug("Cleaning up the resources") #Clean up, terminate the created network offerings @@ -202,7 +67,87 @@ class TestAddNetScaler(cloudstackTestCase): self.debug("Cleanup complete!") except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) - return + + @classmethod + def _addNetScaler(cls, netscaler_config): + """Helper method for common Netscaler add procedure""" + physical_networks = PhysicalNetwork.list( + cls.api_client, + zoneid=cls.zone.id + ) + if isinstance(physical_networks, list): + cls.physical_network = physical_networks[0] + + # Check if a NetScaler network service provider exists - if not add one + nw_service_providers = NetworkServiceProvider.list( + cls.api_client, + name='Netscaler', + physicalnetworkid=cls.physical_network.id + ) + if not isinstance(nw_service_providers, list) or len(nw_service_providers) < 1: + NetworkServiceProvider.add(cls.api_client, + name='Netscaler', + physicalnetworkid=cls.physical_network.id, + servicelist=["Lb"] + ) + + cls.netscaler = NetScaler.add( + cls.api_client, + netscaler_config, + physicalnetworkid=cls.physical_network.id + ) + + nw_service_providers = NetworkServiceProvider.list( + cls.api_client, + name='Netscaler', + physicalnetworkid=cls.physical_network.id + ) + if isinstance(nw_service_providers, list): + cls.netscaler_provider = nw_service_providers[0] + + if cls.netscaler_provider.state != 'Enabled': + NetworkServiceProvider.update( + cls.api_client, + id=cls.netscaler_provider.id, + state='Enabled' + ) + +class _NetScalerAddBase(_NetScalerBase): + """Base class for tests that add a NetScaler (valid & invalid) + Provides standard Setup / TearDown methods""" + @classmethod + def setUpClass(cls): + cls.testClient = super(_NetScalerAddBase, cls).getClsTestClient() + cls.api_client = cls.testClient.getApiClient() + cls.services = cls.testClient.getParsedTestDataConfig() + + # Get Zone, Domain and templates + cls.domain = get_domain(cls.api_client) + cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) + + physical_networks = PhysicalNetwork.list( + cls.api_client, + zoneid=cls.zone.id + ) + assert isinstance(physical_networks, list), "There should be atleast one physical network for advanced zone" + cls.physical_network = physical_networks[0] + + # Check if a NetScaler network service provider exists - if not add one + nw_service_providers = NetworkServiceProvider.list( + cls.api_client, + name='Netscaler', + physicalnetworkid=cls.physical_network.id + ) + if not isinstance(nw_service_providers, list) or len(nw_service_providers) < 1: + NetworkServiceProvider.add(cls.api_client, + name='Netscaler', + physicalnetworkid=cls.physical_network.id, + servicelist=["Lb"] + ) + + cls._cleanup = [] + +class TestAddNetScaler(_NetScalerAddBase): @attr(tags=["advancedns"], required_hardware="true") def test_add_netscaler_device(self): @@ -213,23 +158,14 @@ class TestAddNetScaler(cloudstackTestCase): # credentials , public , private interface and enabling Load # Balancing feature. # 2. Netscaler should be configured successfully. - - physical_networks = PhysicalNetwork.list( - self.apiclient, - zoneid=self.zone.id - ) - self.assertEqual( - isinstance(physical_networks, list), - True, - "There should be atleast one physical network for advanced zone" - ) - physical_network = physical_networks[0] self.debug("Adding netscaler device: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler_VPX"]["ipaddress"]) + netscaler_config = dict(self.services["netscaler_VPX"]) + netscaler_config.update({'lbdevicededicated': "False"}) netscaler = NetScaler.add( self.apiclient, - self.services["netscaler"], - physicalnetworkid=physical_network.id + netscaler_config, + physicalnetworkid=self.physical_network.id ) self.cleanup.append(netscaler) self.debug("Checking if Netscaler network service provider is enabled?") @@ -237,7 +173,7 @@ class TestAddNetScaler(cloudstackTestCase): nw_service_providers = NetworkServiceProvider.list( self.apiclient, name='Netscaler', - physicalnetworkid=physical_network.id + physicalnetworkid=self.physical_network.id ) self.assertEqual( isinstance(nw_service_providers, list), @@ -283,51 +219,11 @@ class TestAddNetScaler(cloudstackTestCase): ) self.assertEqual( ns.physicalnetworkid, - physical_network.id, + self.physical_network.id, "Physical network id should match with the network in which device is configured" ) - return - - -class TestInvalidParametersNetscaler(cloudstackTestCase): - - @classmethod - def setUpClass(cls): - cls.testClient = super(TestInvalidParametersNetscaler, cls).getClsTestClient() - cls.api_client = cls.testClient.getApiClient() - - 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._cleanup = [] - return - - @classmethod - def tearDownClass(cls): - try: - #Cleanup resources used - cleanup_resources(cls.api_client, cls._cleanup) - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return - - def setUp(self): - self.apiclient = self.testClient.getApiClient() - self.dbclient = self.testClient.getDbConnection() - self.cleanup = [] - return - - def tearDown(self): - try: - self.debug("Cleaning up the resources") - #Clean up, terminate the created network offerings - cleanup_resources(self.apiclient, self.cleanup) - self.debug("Cleanup complete!") - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return +class TestInvalidParametersNetscaler(_NetScalerAddBase): @attr(tags=["advancedns"], required_hardware="true") def test_invalid_cred(self): @@ -339,64 +235,23 @@ class TestInvalidParametersNetscaler(cloudstackTestCase): # credentials , but valid public, private interface # 2. Netscaler API should throw error - physical_networks = PhysicalNetwork.list( - self.apiclient, - zoneid=self.zone.id - ) - self.assertEqual( - isinstance(physical_networks, list), - True, - "There should be atleast one physical network for advanced zone" - ) - physical_network = physical_networks[0] - - self.debug("Checking if Netscaler network service provider is enabled?") - - nw_service_providers = NetworkServiceProvider.list( - self.apiclient, - name='Netscaler', - physicalnetworkid=physical_network.id - ) - self.assertEqual( - isinstance(nw_service_providers, list), - True, - "Network service providers list should not be empty" - ) - netscaler_provider = nw_service_providers[0] - - if netscaler_provider.state != 'Enabled': - self.debug("Netscaler provider is not enabled. Enabling it..") - response = NetworkServiceProvider.update( - self.apiclient, - id=netscaler_provider.id, - state='Enabled' - ) - self.assertEqual( - response.state, - "Enabled", - "Network service provider should be in enabled state" - ) - else: - self.debug("NetScaler service provider is already enabled.") - self.debug("Passing invalid credential for NetScaler") - self.services["netscaler"]["username"] = random_gen() - self.services["netscaler"]["password"] = random_gen() + netscaler_config = dict(self.services["netscaler_VPX"]) + netscaler_config.update({'username': random_gen(), 'password': random_gen()}) self.debug("Adding netscaler device: %s" % - self.services["netscaler"]["ipaddress"]) + netscaler_config["ipaddress"]) self.debug("Username: %s, password: %s" % ( - self.services["netscaler"]["username"], - self.services["netscaler"]["password"] + netscaler_config["username"], + netscaler_config["password"] )) with self.assertRaises(Exception): NetScaler.add( - self.apiclient, - self.services["netscaler"], - physicalnetworkid=physical_network.id - ) - return + self.apiclient, + netscaler_config, + physicalnetworkid=self.physical_network.id + ) @attr(tags=["advancedns"], required_hardware="true") def test_invalid_public_interface(self): @@ -408,62 +263,21 @@ class TestInvalidParametersNetscaler(cloudstackTestCase): # credentials , private interface and invalid public interface # 2. Netscaler API should throw error - physical_networks = PhysicalNetwork.list( - self.apiclient, - zoneid=self.zone.id - ) - self.assertEqual( - isinstance(physical_networks, list), - True, - "There should be atleast one physical network for advanced zone" - ) - physical_network = physical_networks[0] - - self.debug("Checking if Netscaler network service provider is enabled?") - - nw_service_providers = NetworkServiceProvider.list( - self.apiclient, - name='Netscaler', - physicalnetworkid=physical_network.id - ) - self.assertEqual( - isinstance(nw_service_providers, list), - True, - "Network service providers list should not be empty" - ) - netscaler_provider = nw_service_providers[0] - - if netscaler_provider.state != 'Enabled': - self.debug("Netscaler provider is not enabled. Enabling it..") - response = NetworkServiceProvider.update( - self.apiclient, - id=netscaler_provider.id, - state='Enabled' - ) - self.assertEqual( - response.state, - "Enabled", - "Network service provider should be in enabled state" - ) - else: - self.debug("NetScaler service provider is already enabled.") - self.debug("Passing invalid public interface for NetScaler") - self.services["netscaler"]["publicinterface"] = random_gen() - + netscaler_config = dict(self.services["netscaler_VPX"]) + netscaler_config.update({'publicinterface': random_gen()}) self.debug("Adding netscaler device: %s" % - self.services["netscaler"]["ipaddress"]) + netscaler_config["ipaddress"]) self.debug("Public interface: %s" % - self.services["netscaler"]["publicinterface"]) + netscaler_config["publicinterface"]) with self.assertRaises(Exception): NetScaler.add( - self.apiclient, - self.services["netscaler"], - physicalnetworkid=physical_network.id - ) - return + self.apiclient, + netscaler_config, + physicalnetworkid=self.physical_network.id + ) @attr(tags=["advancedns"], required_hardware="true") def test_invalid_private_interface(self): @@ -474,142 +288,24 @@ class TestInvalidParametersNetscaler(cloudstackTestCase): # 1. Add Netscaler device into a Zone by providing valid log in # credentials , public interface and invalid private interface # 2. Netscaler API should throw error - - physical_networks = PhysicalNetwork.list( - self.apiclient, - zoneid=self.zone.id - ) - self.assertEqual( - isinstance(physical_networks, list), - True, - "There should be atleast one physical network for advanced zone" - ) - physical_network = physical_networks[0] - - self.debug("Checking if Netscaler network service provider is enabled?") - - nw_service_providers = NetworkServiceProvider.list( - self.apiclient, - name='Netscaler', - physicalnetworkid=physical_network.id - ) - self.assertEqual( - isinstance(nw_service_providers, list), - True, - "Network service providers list should not be empty" - ) - netscaler_provider = nw_service_providers[0] - - if netscaler_provider.state != 'Enabled': - self.debug("Netscaler provider is not enabled. Enabling it..") - response = NetworkServiceProvider.update( - self.apiclient, - id=netscaler_provider.id, - state='Enabled' - ) - self.assertEqual( - response.state, - "Enabled", - "Network service provider should be in enabled state" - ) - else: - self.debug("NetScaler service provider is already enabled.") - - self.debug("Passing invalid private interface for NetScaler") - self.services["netscaler"]["privateinterface"] = random_gen() - + netscaler_config = dict(self.services["netscaler_VPX"]) + netscaler_config.update({'privateinterface': random_gen()}) self.debug("Adding netscaler device: %s" % - self.services["netscaler"]["ipaddress"]) + netscaler_config["ipaddress"]) self.debug("Private interface: %s" % - self.services["netscaler"]["privateinterface"]) + netscaler_config["privateinterface"]) with self.assertRaises(Exception): NetScaler.add( - self.apiclient, - self.services["netscaler"], - physicalnetworkid=physical_network.id - ) - return + self.apiclient, + netscaler_config, + physicalnetworkid=self.physical_network.id + ) - -class TestNetScalerDedicated(cloudstackTestCase): - - @classmethod - def setUpClass(cls): - cls.testClient = super(TestNetScalerDedicated, cls).getClsTestClient() - cls.api_client = cls.testClient.getApiClient() - - 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.template = get_template( - cls.api_client, - cls.zone.id, - cls.services["ostype"] - ) - - physical_networks = PhysicalNetwork.list( - cls.api_client, - zoneid=cls.zone.id - ) - if isinstance(physical_networks, list): - physical_network = physical_networks[0] - - cls.netscaler = NetScaler.add( - cls.api_client, - cls.services["netscaler_dedicated"], - physicalnetworkid=physical_network.id - ) - - nw_service_providers = NetworkServiceProvider.list( - cls.api_client, - name='Netscaler', - physicalnetworkid=physical_network.id - ) - if isinstance(nw_service_providers, list): - netscaler_provider = nw_service_providers[0] - - if netscaler_provider.state != 'Enabled': - NetworkServiceProvider.update( - cls.api_client, - id=netscaler_provider.id, - state='Enabled' - ) - - cls.network_offering = NetworkOffering.create( - cls.api_client, - cls.services["network_offering_dedicated"], - conservemode=False, - state="Creating", - status="Creating", - allocationstate="Creating", - ) - # Enable Network offering - cls.network_offering.update(cls.api_client, state='Enabled') - 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._cleanup = [ - cls.service_offering, - cls.network_offering, - cls.netscaler, - ] - return - - @classmethod - def tearDownClass(cls): - try: - #Cleanup resources used - cleanup_resources(cls.api_client, cls._cleanup) - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return +class _NetScalerDeployVMBase(_NetScalerBase): + """Base class for testing VM deployment using NetScaler networking + Provides standard test-case setup / tear-down methods""" def setUp(self): self.apiclient = self.testClient.getApiClient() @@ -627,7 +323,6 @@ class TestNetScalerDedicated(cloudstackTestCase): domainid=self.domain.id ) self.cleanup = [self.account_1, self.account_2] - return def tearDown(self): try: @@ -647,7 +342,56 @@ class TestNetScalerDedicated(cloudstackTestCase): self.debug("Cleanup complete!") except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) - return + +class TestNetScalerDedicated(_NetScalerDeployVMBase): + + @classmethod + def setUpClass(cls): + cls.testClient = super(TestNetScalerDedicated, cls).getClsTestClient() + cls.api_client = cls.testClient.getApiClient() + cls.services = cls.testClient.getParsedTestDataConfig() + + # Get Zone, Domain and templates + 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"] + ) + + netscaler_config = dict(cls.services["netscaler_VPX"]) + netscaler_config.update({'lbdevicededicated': "True"}) + cls._addNetScaler(netscaler_config) + + nw_offering_config = dict(cls.services["nw_off_isolated_netscaler"]) + nw_offering_config.update({"serviceCapabilityList": { + "SourceNat": { + "SupportedSourceNatTypes": "peraccount" + }, + "lb": { + "SupportedLbIsolation": "dedicated" + } + } + }) + + cls.network_offering = NetworkOffering.create( + cls.api_client, + nw_offering_config, + conservemode=False + ) + # Enable Network offering + cls.network_offering.update(cls.api_client, state='Enabled') + + cls.service_offering = ServiceOffering.create( + cls.api_client, + cls.services["service_offering"] + ) + cls._cleanup = [ + cls.service_offering, + cls.network_offering, + cls.netscaler, + ] @attr(tags = ["advancedns"]) def test_netscaler_dedicated_mode(self): @@ -685,7 +429,7 @@ class TestNetScalerDedicated(cloudstackTestCase): self.network_offering.id) self.network = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_1.name, domainid=self.account_1.domainid, networkofferingid=self.network_offering.id, @@ -702,7 +446,9 @@ class TestNetScalerDedicated(cloudstackTestCase): accountid=self.account_1.name, domainid=self.account_1.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network.id)] + networkids=[str(self.network.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network.id) list_vm_response = VirtualMachine.list( @@ -734,7 +480,7 @@ class TestNetScalerDedicated(cloudstackTestCase): Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_2.name, domainid=self.account_2.domainid, networkofferingid=self.network_offering.id, @@ -748,21 +494,20 @@ class TestNetScalerDedicated(cloudstackTestCase): accountid=self.account_2.name, domainid=self.account_2.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network.id)] + networkids=[str(self.network.id)], + templateid=self.template.id, + zoneid=self.zone.id ) - self.debug("Deply instance in dedicated Network offering mode failed") - return + self.debug("Deply instance in dedicated Network offering mode failed") - - -class TestNetScalerShared(cloudstackTestCase): +class TestNetScalerShared(_NetScalerDeployVMBase): @classmethod def setUpClass(cls): cls.testClient = super(TestNetScalerShared, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() + cls.services = cls.testClient.getParsedTestDataConfig() - 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()) @@ -772,42 +517,17 @@ class TestNetScalerShared(cloudstackTestCase): cls.services["ostype"] ) - physical_networks = PhysicalNetwork.list( - cls.api_client, - zoneid=cls.zone.id - ) - if isinstance(physical_networks, list): - physical_network = physical_networks[0] + netscaler_config = dict(cls.services["netscaler_VPX"]) + netscaler_config.update({'lbdevicededicated': "False"}) + cls._addNetScaler(netscaler_config) - cls.netscaler = NetScaler.add( - cls.api_client, - cls.services["netscaler"], - physicalnetworkid=physical_network.id - ) - - nw_service_providers = NetworkServiceProvider.list( - cls.api_client, - name='Netscaler', - physicalnetworkid=physical_network.id - ) - if isinstance(nw_service_providers, list): - netscaler_provider = nw_service_providers[0] - - if netscaler_provider.state != 'Enabled': - NetworkServiceProvider.update( - cls.api_client, - id=netscaler_provider.id, - state='Enabled' - ) cls.network_offering = NetworkOffering.create( cls.api_client, - cls.services["network_offering"], + cls.services["nw_off_isolated_netscaler"], conservemode=True ) # Enable Network offering cls.network_offering.update(cls.api_client, state='Enabled') - cls.services["virtual_machine"]["zoneid"] = cls.zone.id - cls.services["virtual_machine"]["template"] = cls.template.id cls.service_offering = ServiceOffering.create( cls.api_client, @@ -818,54 +538,6 @@ class TestNetScalerShared(cloudstackTestCase): cls.network_offering, cls.netscaler, ] - return - - @classmethod - def tearDownClass(cls): - try: - #Cleanup resources used - cleanup_resources(cls.api_client, cls._cleanup) - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return - - def setUp(self): - self.apiclient = self.testClient.getApiClient() - self.dbclient = self.testClient.getDbConnection() - self.account_1 = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) - self.account_2 = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) - self.cleanup = [self.account_1, self.account_2] - return - - def tearDown(self): - try: - self.debug("Cleaning up the resources") - #Clean up, terminate the created network offerings - cleanup_resources(self.apiclient, self.cleanup) - interval = Configurations.list( - self.apiclient, - name='network.gc.interval' - ) - wait = Configurations.list( - self.apiclient, - name='network.gc.wait' - ) - # Sleep to ensure that all resources are deleted - time.sleep(int(interval[0].value) + int(wait[0].value)) - self.debug("Cleanup complete!") - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return @attr(tags = ["advancedns"]) def test_netscaler_shared_mode(self): @@ -903,7 +575,7 @@ class TestNetScalerShared(cloudstackTestCase): self.network_offering.id) self.network_1 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_1.name, domainid=self.account_1.domainid, networkofferingid=self.network_offering.id, @@ -920,7 +592,9 @@ class TestNetScalerShared(cloudstackTestCase): accountid=self.account_1.name, domainid=self.account_1.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_1.id)] + networkids=[str(self.network_1.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network_1.id) list_vm_response = VirtualMachine.list( @@ -951,7 +625,7 @@ class TestNetScalerShared(cloudstackTestCase): self.network_offering.id) self.network_2 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_2.name, domainid=self.account_2.domainid, networkofferingid=self.network_offering.id, @@ -968,7 +642,9 @@ class TestNetScalerShared(cloudstackTestCase): accountid=self.account_2.name, domainid=self.account_2.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_2.id)] + networkids=[str(self.network_2.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network_2.id) list_vm_response = VirtualMachine.list( @@ -993,84 +669,10 @@ class TestNetScalerShared(cloudstackTestCase): "Running", "VM state should be running after deployment" ) - return - - - -class TestNetScalerCustomCapacity(cloudstackTestCase): - - @classmethod - def setUpClass(cls): - cls.testClient = super(TestNetScalerCustomCapacity, cls).getClsTestClient() - cls.api_client = cls.testClient.getApiClient() - - 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.template = get_template( - cls.api_client, - cls.zone.id, - cls.services["ostype"] - ) - - physical_networks = PhysicalNetwork.list( - cls.api_client, - zoneid=cls.zone.id - ) - if isinstance(physical_networks, list): - physical_network = physical_networks[0] - cls.services["netscaler"]["lbdevicecapacity"] = 2 - cls.netscaler = NetScaler.add( - cls.api_client, - cls.services["netscaler"], - physicalnetworkid=physical_network.id - ) - - nw_service_providers = NetworkServiceProvider.list( - cls.api_client, - name='Netscaler', - physicalnetworkid=physical_network.id - ) - if isinstance(nw_service_providers, list): - netscaler_provider = nw_service_providers[0] - - if netscaler_provider.state != 'Enabled': - NetworkServiceProvider.update( - cls.api_client, - id=netscaler_provider.id, - state='Enabled' - ) - cls.network_offering = NetworkOffering.create( - cls.api_client, - cls.services["network_offering"], - conservemode=True - ) - # Enable Network offering - cls.network_offering.update(cls.api_client, state='Enabled') - 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._cleanup = [ - cls.service_offering, - cls.network_offering, - cls.netscaler, - ] - return - - @classmethod - def tearDownClass(cls): - try: - #Cleanup resources used - cleanup_resources(cls.api_client, cls._cleanup) - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return +class _NetScalerCapacity(_NetScalerDeployVMBase): + """Base class for NetScaler capacity tests + Provides standard test-case setup / tear-down method""" def setUp(self): self.apiclient = self.testClient.getApiClient() self.dbclient = self.testClient.getDbConnection() @@ -1093,27 +695,45 @@ class TestNetScalerCustomCapacity(cloudstackTestCase): domainid=self.domain.id ) self.cleanup = [self.account_1, self.account_2, self.account_3] - return - def tearDown(self): - try: - self.debug("Cleaning up the resources") - #Clean up, terminate the created network offerings - cleanup_resources(self.apiclient, self.cleanup) - interval = Configurations.list( - self.apiclient, - name='network.gc.interval' - ) - wait = Configurations.list( - self.apiclient, - name='network.gc.wait' - ) - # Sleep to ensure that all resources are deleted - time.sleep(int(interval[0].value) + int(wait[0].value)) - self.debug("Cleanup complete!") - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return +class TestNetScalerCustomCapacity(_NetScalerCapacity): + + @classmethod + def setUpClass(cls): + cls.testClient = super(TestNetScalerCustomCapacity, cls).getClsTestClient() + cls.api_client = cls.testClient.getApiClient() + cls.services = cls.testClient.getParsedTestDataConfig() + + # Get Zone, Domain and templates + 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"] + ) + + netscaler_config = dict(cls.services["netscaler_VPX"]) + netscaler_config.update({'lbdevicecapacity': 3, 'lbdevicededicated': "False"}) + cls._addNetScaler(netscaler_config) + + cls.network_offering = NetworkOffering.create( + cls.api_client, + cls.services["nw_off_isolated_netscaler"], + conservemode=True + ) + # Enable Network offering + cls.network_offering.update(cls.api_client, state='Enabled') + + cls.service_offering = ServiceOffering.create( + cls.api_client, + cls.services["service_offering"] + ) + cls._cleanup = [ + cls.service_offering, + cls.network_offering, + cls.netscaler, + ] @attr(tags = ["advancedns","test"]) def test_netscaler_custom_capacity(self): @@ -1151,7 +771,7 @@ class TestNetScalerCustomCapacity(cloudstackTestCase): self.network_offering.id) self.network_1 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_1.name, domainid=self.account_1.domainid, networkofferingid=self.network_offering.id, @@ -1168,7 +788,9 @@ class TestNetScalerCustomCapacity(cloudstackTestCase): accountid=self.account_1.name, domainid=self.account_1.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_1.id)] + networkids=[str(self.network_1.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network_1.id) list_vm_response = VirtualMachine.list( @@ -1199,7 +821,7 @@ class TestNetScalerCustomCapacity(cloudstackTestCase): self.network_offering.id) self.network_2 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_2.name, domainid=self.account_2.domainid, networkofferingid=self.network_offering.id, @@ -1216,7 +838,9 @@ class TestNetScalerCustomCapacity(cloudstackTestCase): accountid=self.account_2.name, domainid=self.account_2.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_2.id)] + networkids=[str(self.network_2.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network_2.id) list_vm_response = VirtualMachine.list( @@ -1246,7 +870,7 @@ class TestNetScalerCustomCapacity(cloudstackTestCase): self.network_offering.id) self.network_3 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_3.name, domainid=self.account_3.domainid, networkofferingid=self.network_offering.id, @@ -1255,28 +879,50 @@ class TestNetScalerCustomCapacity(cloudstackTestCase): self.debug("Created network with ID: %s" % self.network_3.id) self.debug("Deploying VM in account: %s" % self.account_3.name) - with self.assertRaises(Exception): - # Spawn an instance in that network - VirtualMachine.create( + # Spawn an instance in that network + virtual_machine_3 = VirtualMachine.create( self.apiclient, self.services["virtual_machine"], accountid=self.account_3.name, domainid=self.account_3.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_3.id)] + networkids=[str(self.network_3.id)], + templateid=self.template.id, + zoneid=self.zone.id ) - return + self.debug("Deployed VM in network: %s" % self.network_3.id) + list_vm_response = VirtualMachine.list( + self.apiclient, + id=virtual_machine_3.id + ) + self.debug( + "Verify listVirtualMachines response for virtual machine: %s" \ + % virtual_machine_3.id + ) -class TestNetScalerNoCapacity(cloudstackTestCase): + self.assertEqual( + isinstance(list_vm_response, list), + True, + "Check list response returns a valid list" + ) + vm_response = list_vm_response[0] + + self.assertEqual( + vm_response.state, + "Running", + "VM state should be running after deployment" + ) + +class TestNetScalerNoCapacity(_NetScalerCapacity): @classmethod def setUpClass(cls): cls.testClient = super(TestNetScalerNoCapacity, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() + cls.services = cls.testClient.getParsedTestDataConfig() - 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()) @@ -1286,42 +932,17 @@ class TestNetScalerNoCapacity(cloudstackTestCase): cls.services["ostype"] ) - physical_networks = PhysicalNetwork.list( - cls.api_client, - zoneid=cls.zone.id - ) - if isinstance(physical_networks, list): - physical_network = physical_networks[0] - cls.services["netscaler"]["lbdevicecapacity"] = 2 - cls.netscaler = NetScaler.add( - cls.api_client, - cls.services["netscaler"], - physicalnetworkid=physical_network.id - ) + netscaler_config = dict(cls.services["netscaler_VPX"]) + netscaler_config.update({'lbdevicecapacity': 2, 'lbdevicededicated': "False"}) + cls._addNetScaler(netscaler_config) - nw_service_providers = NetworkServiceProvider.list( - cls.api_client, - name='Netscaler', - physicalnetworkid=physical_network.id - ) - if isinstance(nw_service_providers, list): - netscaler_provider = nw_service_providers[0] - - if netscaler_provider.state != 'Enabled': - NetworkServiceProvider.update( - cls.api_client, - id=netscaler_provider.id, - state='Enabled' - ) cls.network_offering = NetworkOffering.create( cls.api_client, - cls.services["network_offering"], + cls.services["nw_off_isolated_netscaler"], conservemode=True ) # Enable Network offering cls.network_offering.update(cls.api_client, state='Enabled') - cls.services["virtual_machine"]["zoneid"] = cls.zone.id - cls.services["virtual_machine"]["template"] = cls.template.id cls.service_offering = ServiceOffering.create( cls.api_client, @@ -1332,60 +953,6 @@ class TestNetScalerNoCapacity(cloudstackTestCase): cls.network_offering, cls.netscaler, ] - return - - @classmethod - def tearDownClass(cls): - try: - #Cleanup resources used - cleanup_resources(cls.api_client, cls._cleanup) - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return - - def setUp(self): - self.apiclient = self.testClient.getApiClient() - self.dbclient = self.testClient.getDbConnection() - self.account_1 = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) - self.account_2 = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) - self.account_3 = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) - self.cleanup = [self.account_1, self.account_2, self.account_3] - return - - def tearDown(self): - try: - self.debug("Cleaning up the resources") - #Clean up, terminate the created network offerings - cleanup_resources(self.apiclient, self.cleanup) - interval = Configurations.list( - self.apiclient, - name='network.gc.interval' - ) - wait = Configurations.list( - self.apiclient, - name='network.gc.wait' - ) - # Sleep to ensure that all resources are deleted - time.sleep(int(interval[0].value) + int(wait[0].value)) - self.debug("Cleanup complete!") - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return @attr(tags = ["advancedns","test"]) def test_netscaler_no_capacity(self): @@ -1424,7 +991,7 @@ class TestNetScalerNoCapacity(cloudstackTestCase): self.network_offering.id) self.network_1 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_1.name, domainid=self.account_1.domainid, networkofferingid=self.network_offering.id, @@ -1441,7 +1008,9 @@ class TestNetScalerNoCapacity(cloudstackTestCase): accountid=self.account_1.name, domainid=self.account_1.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_1.id)] + networkids=[str(self.network_1.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network_1.id) list_vm_response = VirtualMachine.list( @@ -1472,7 +1041,7 @@ class TestNetScalerNoCapacity(cloudstackTestCase): self.network_offering.id) self.network_2 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_2.name, domainid=self.account_2.domainid, networkofferingid=self.network_offering.id, @@ -1489,7 +1058,9 @@ class TestNetScalerNoCapacity(cloudstackTestCase): accountid=self.account_2.name, domainid=self.account_2.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_2.id)] + networkids=[str(self.network_2.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network_2.id) list_vm_response = VirtualMachine.list( @@ -1519,7 +1090,7 @@ class TestNetScalerNoCapacity(cloudstackTestCase): self.network_offering.id) self.network_3 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_3.name, domainid=self.account_3.domainid, networkofferingid=self.network_offering.id, @@ -1536,20 +1107,19 @@ class TestNetScalerNoCapacity(cloudstackTestCase): accountid=self.account_3.name, domainid=self.account_3.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_3.id)] + networkids=[str(self.network_3.id)], + templateid=self.template.id, + zoneid=self.zone.id ) - return - - -class TestGuestNetworkWithNetScaler(cloudstackTestCase): +class TestGuestNetworkWithNetScaler(_NetScalerDeployVMBase): @classmethod def setUpClass(cls): cls.testClient = super(TestGuestNetworkWithNetScaler, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() + cls.services = cls.testClient.getParsedTestDataConfig() - 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()) @@ -1559,42 +1129,17 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): cls.services["ostype"] ) - physical_networks = PhysicalNetwork.list( - cls.api_client, - zoneid=cls.zone.id - ) - if isinstance(physical_networks, list): - physical_network = physical_networks[0] + netscaler_config = dict(cls.services["netscaler_VPX"]) + netscaler_config.update({'lbdevicededicated': "False"}) + cls._addNetScaler(netscaler_config) - cls.netscaler = NetScaler.add( - cls.api_client, - cls.services["netscaler"], - physicalnetworkid=physical_network.id - ) - - nw_service_providers = NetworkServiceProvider.list( - cls.api_client, - name='Netscaler', - physicalnetworkid=physical_network.id - ) - if isinstance(nw_service_providers, list): - netscaler_provider = nw_service_providers[0] - - if netscaler_provider.state != 'Enabled': - NetworkServiceProvider.update( - cls.api_client, - id=netscaler_provider.id, - state='Enabled' - ) cls.network_offering = NetworkOffering.create( cls.api_client, - cls.services["network_offering"], + cls.services["nw_off_isolated_netscaler"], conservemode=True ) # Enable Network offering cls.network_offering.update(cls.api_client, state='Enabled') - cls.services["virtual_machine"]["zoneid"] = cls.zone.id - cls.services["virtual_machine"]["template"] = cls.template.id cls.service_offering = ServiceOffering.create( cls.api_client, @@ -1605,54 +1150,7 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): cls.network_offering, cls.netscaler, ] - return - @classmethod - def tearDownClass(cls): - try: - #Cleanup resources used - cleanup_resources(cls.api_client, cls._cleanup) - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return - - def setUp(self): - self.apiclient = self.testClient.getApiClient() - self.dbclient = self.testClient.getDbConnection() - self.account_1 = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) - self.account_2 = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) - self.cleanup = [self.account_1, self.account_2] - return - - def tearDown(self): - try: - self.debug("Cleaning up the resources") - #Clean up, terminate the created network offerings - cleanup_resources(self.apiclient, self.cleanup) - interval = Configurations.list( - self.apiclient, - name='network.gc.interval' - ) - wait = Configurations.list( - self.apiclient, - name='network.gc.wait' - ) - # Sleep to ensure that all resources are deleted - time.sleep(int(interval[0].value) + int(wait[0].value)) - self.debug("Cleanup complete!") - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return @attr(tags = ["advancedns","test"]) def test_01_guest_network(self): @@ -1699,7 +1197,7 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): self.network_offering.id) self.network_1 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_1.name, domainid=self.account_1.domainid, networkofferingid=self.network_offering.id, @@ -1716,7 +1214,9 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): accountid=self.account_1.name, domainid=self.account_1.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_1.id)] + networkids=[str(self.network_1.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network_1.id) list_vm_response = VirtualMachine.list( @@ -1749,13 +1249,13 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): ) nw = network_list[0] self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler_VPX"]["ipaddress"]) try: ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - self.services["netscaler"]["port"], - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], + self.services["netscaler_VPX"]["ipaddress"], + self.services["netscaler_VPX"]["port"], + self.services["netscaler_VPX"]["username"], + self.services["netscaler_VPX"]["password"], ) cmd = "show vlan %s" % (nw.vlan) self.debug("command: %s" % cmd) @@ -1771,7 +1271,7 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): ) except Exception as e: self.fail("SSH Access failed for %s: %s" % \ - (self.services["netscaler"]["ipaddress"], e)) + (self.services["netscaler_VPX"]["ipaddress"], e)) return @attr(tags = ["advancedns","test"]) @@ -1798,7 +1298,7 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): self.network_offering.id) self.network_1 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_1.name, domainid=self.account_1.domainid, networkofferingid=self.network_offering.id, @@ -1815,7 +1315,9 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): accountid=self.account_1.name, domainid=self.account_1.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_1.id)] + networkids=[str(self.network_1.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network_1.id) list_vm_response = VirtualMachine.list( @@ -1845,7 +1347,7 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): self.network_offering.id) self.network_2 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_2.name, domainid=self.account_2.domainid, networkofferingid=self.network_offering.id, @@ -1862,7 +1364,9 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): accountid=self.account_2.name, domainid=self.account_2.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_2.id)] + networkids=[str(self.network_2.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network_2.id) list_vm_response = VirtualMachine.list( @@ -1888,7 +1392,7 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): "VM state should be running after deployment" ) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler_VPX"]["ipaddress"]) try: # Find Network vlan used network_list = Network.list( @@ -1898,10 +1402,10 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): ) nw = network_list[0] ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - self.services["netscaler"]["port"], - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], + self.services["netscaler_VPX"]["ipaddress"], + self.services["netscaler_VPX"]["port"], + self.services["netscaler_VPX"]["username"], + self.services["netscaler_VPX"]["password"], ) cmd = "show vlan %s" % (nw.vlan) self.debug("command: %s" % cmd) @@ -1937,7 +1441,7 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): ) except Exception as e: self.fail("SSH Access failed for %s: %s" % \ - (self.services["netscaler"]["ipaddress"], e)) + (self.services["netscaler_VPX"]["ipaddress"], e)) return @attr(tags = ["advancedns","test"]) @@ -1962,7 +1466,7 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): self.network_offering.id) self.network = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_1.name, domainid=self.account_1.domainid, networkofferingid=self.network_offering.id, @@ -1979,7 +1483,9 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): accountid=self.account_1.name, domainid=self.account_1.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network.id)] + networkids=[str(self.network.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network.id) list_vm_response = VirtualMachine.list( @@ -2055,13 +1561,13 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): time.sleep(int(interval[0].value) + int(wait[0].value)) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler_VPX"]["ipaddress"]) try: ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - self.services["netscaler"]["port"], - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], + self.services["netscaler_VPX"]["ipaddress"], + self.services["netscaler_VPX"]["port"], + self.services["netscaler_VPX"]["username"], + self.services["netscaler_VPX"]["password"], ) cmd = "show vlan %s" % (nw.vlan) self.debug("command: %s" % cmd) @@ -2092,20 +1598,20 @@ class TestGuestNetworkWithNetScaler(cloudstackTestCase): ) except Exception as e: self.fail("SSH Access failed for %s: %s" % \ - (self.services["netscaler"]["ipaddress"], e)) + (self.services["netscaler_VPX"]["ipaddress"], e)) return -class TestGuestNetworkShutDown(cloudstackTestCase): +class TestGuestNetworkShutDown(_NetScalerBase): @classmethod def setUpClass(cls): cls._cleanup = [] cls.testClient = super(TestGuestNetworkShutDown, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() + cls.services = cls.testClient.getParsedTestDataConfig() - 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()) @@ -2115,17 +1621,17 @@ class TestGuestNetworkShutDown(cloudstackTestCase): cls.services["ostype"] ) try: - cls.netscaler = add_netscaler(cls.api_client, cls.zone.id, cls.services) + netscaler_config = dict(cls.services["netscaler_VPX"]) + netscaler_config.update({'lbdevicededicated': "False"}) + cls.netscaler = add_netscaler(cls.api_client, cls.zone.id, netscaler_config) cls._cleanup.append(cls.netscaler) cls.network_offering = NetworkOffering.create( cls.api_client, - cls.services["network_offering"], + cls.services["nw_off_isolated_netscaler"], conservemode=True ) # Enable Network offering cls.network_offering.update(cls.api_client, state='Enabled') - cls.services["virtual_machine"]["zoneid"] = cls.zone.id - cls.services["virtual_machine"]["template"] = cls.template.id cls.service_offering = ServiceOffering.create( cls.api_client, @@ -2141,7 +1647,7 @@ class TestGuestNetworkShutDown(cloudstackTestCase): # Creating network using the network offering created cls.network = Network.create( cls.api_client, - cls.services["network"], + cls.services["netscaler_network"], accountid=cls.account.name, domainid=cls.account.domainid, networkofferingid=cls.network_offering.id, @@ -2155,7 +1661,9 @@ class TestGuestNetworkShutDown(cloudstackTestCase): accountid=cls.account.name, domainid=cls.account.domainid, serviceofferingid=cls.service_offering.id, - networkids=[str(cls.network.id)] + networkids=[str(cls.network.id)], + templateid=cls.template.id, + zoneid=cls.zone.id ) cls.vm_2 = VirtualMachine.create( cls.api_client, @@ -2163,7 +1671,9 @@ class TestGuestNetworkShutDown(cloudstackTestCase): accountid=cls.account.name, domainid=cls.account.domainid, serviceofferingid=cls.service_offering.id, - networkids=[str(cls.network.id)] + networkids=[str(cls.network.id)], + templateid=cls.template.id, + zoneid=cls.zone.id ) cls.public_ip = PublicIPAddress.create( cls.api_client, @@ -2184,33 +1694,6 @@ class TestGuestNetworkShutDown(cloudstackTestCase): cls.tearDownClass() raise Exception ("Warning: Exception in setUpClass: %s" % e) - return - - @classmethod - def tearDownClass(cls): - try: - #Cleanup resources used - cleanup_resources(cls.api_client, cls._cleanup) - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return - - def setUp(self): - self.apiclient = self.testClient.getApiClient() - self.dbclient = self.testClient.getDbConnection() - self.cleanup = [] - return - - def tearDown(self): - try: - self.debug("Cleaning up the resources") - #Clean up, terminate the created network offerings - cleanup_resources(self.apiclient, self.cleanup) - self.debug("Cleanup complete!") - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return - @attr(tags = ["advancedns","test"]) def test_01_stop_all_vms(self): """Test Stopping all the Vms for any account that has LB rules. @@ -2255,13 +1738,13 @@ class TestGuestNetworkShutDown(cloudstackTestCase): time.sleep((int(interval[0].value) + int(wait[0].value)) * 2) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler_VPX"]["ipaddress"]) try: ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - self.services["netscaler"]["port"], - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], + self.services["netscaler_VPX"]["ipaddress"], + self.services["netscaler_VPX"]["port"], + self.services["netscaler_VPX"]["username"], + self.services["netscaler_VPX"]["password"], ) cmd = "show vlan %s" % (nw.vlan) self.debug("command: %s" % cmd) @@ -2292,7 +1775,7 @@ class TestGuestNetworkShutDown(cloudstackTestCase): ) except Exception as e: self.fail("SSH Access failed for %s: %s" % \ - (self.services["netscaler"]["ipaddress"], e)) + (self.services["netscaler_VPX"]["ipaddress"], e)) return @attr(tags = ["advancedns","test"]) @@ -2333,7 +1816,7 @@ class TestGuestNetworkShutDown(cloudstackTestCase): ) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler_VPX"]["ipaddress"]) try: # Find Network vlan used network_list = Network.list( @@ -2343,10 +1826,10 @@ class TestGuestNetworkShutDown(cloudstackTestCase): ) nw = network_list[0] ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - self.services["netscaler"]["port"], - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], + self.services["netscaler_VPX"]["ipaddress"], + self.services["netscaler_VPX"]["port"], + self.services["netscaler_VPX"]["username"], + self.services["netscaler_VPX"]["password"], ) cmd = "show vlan %s" % (nw.vlan) self.debug("command: %s" % cmd) @@ -2377,7 +1860,7 @@ class TestGuestNetworkShutDown(cloudstackTestCase): ) except Exception as e: self.fail("SSH Access failed for %s: %s" % \ - (self.services["netscaler"]["ipaddress"], e)) + (self.services["netscaler_VPX"]["ipaddress"], e)) return @attr(tags = ["advancedns","test"]) @@ -2395,7 +1878,7 @@ class TestGuestNetworkShutDown(cloudstackTestCase): self.network.restart(self.apiclient, cleanup=False) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler_VPX"]["ipaddress"]) try: # Find Network vlan used network_list = Network.list( @@ -2405,10 +1888,10 @@ class TestGuestNetworkShutDown(cloudstackTestCase): ) nw = network_list[0] ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - self.services["netscaler"]["port"], - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], + self.services["netscaler_VPX"]["ipaddress"], + self.services["netscaler_VPX"]["port"], + self.services["netscaler_VPX"]["username"], + self.services["netscaler_VPX"]["password"], ) cmd = "show vlan %s" % (nw.vlan) self.debug("command: %s" % cmd) @@ -2439,7 +1922,7 @@ class TestGuestNetworkShutDown(cloudstackTestCase): ) except Exception as e: self.fail("SSH Access failed for %s: %s" % \ - (self.services["netscaler"]["ipaddress"], e)) + (self.services["netscaler_VPX"]["ipaddress"], e)) return @attr(tags = ["advancedns","test"]) @@ -2457,7 +1940,7 @@ class TestGuestNetworkShutDown(cloudstackTestCase): self.network.restart(self.apiclient, cleanup=True) self.debug("SSH into netscaler: %s" % - self.services["netscaler"]["ipaddress"]) + self.services["netscaler_VPX"]["ipaddress"]) try: # Find Network vlan used network_list = Network.list( @@ -2467,10 +1950,10 @@ class TestGuestNetworkShutDown(cloudstackTestCase): ) nw = network_list[0] ssh_client = SshClient( - self.services["netscaler"]["ipaddress"], - self.services["netscaler"]["port"], - self.services["netscaler"]["username"], - self.services["netscaler"]["password"], + self.services["netscaler_VPX"]["ipaddress"], + self.services["netscaler_VPX"]["port"], + self.services["netscaler_VPX"]["username"], + self.services["netscaler_VPX"]["password"], ) cmd = "show vlan %s" % (nw.vlan) self.debug("command: %s" % cmd) @@ -2501,19 +1984,19 @@ class TestGuestNetworkShutDown(cloudstackTestCase): ) except Exception as e: self.fail("SSH Access failed for %s: %s" % \ - (self.services["netscaler"]["ipaddress"], e)) + (self.services["netscaler_VPX"]["ipaddress"], e)) return -class TestServiceProvider(cloudstackTestCase): +class TestServiceProvider(_NetScalerDeployVMBase): @classmethod def setUpClass(cls): cls.testClient = super(TestServiceProvider, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() + cls.services = cls.testClient.getParsedTestDataConfig() - 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()) @@ -2523,42 +2006,16 @@ class TestServiceProvider(cloudstackTestCase): cls.services["ostype"] ) - physical_networks = PhysicalNetwork.list( - cls.api_client, - zoneid=cls.zone.id - ) - if isinstance(physical_networks, list): - physical_network = physical_networks[0] + netscaler_config = dict(cls.services["netscaler_VPX"]) + netscaler_config.update({'lbdevicededicated': "False"}) + cls._addNetScaler(netscaler_config) - cls.netscaler = NetScaler.add( - cls.api_client, - cls.services["netscaler"], - physicalnetworkid=physical_network.id - ) - - nw_service_providers = NetworkServiceProvider.list( - cls.api_client, - name='Netscaler', - physicalnetworkid=physical_network.id - ) - if isinstance(nw_service_providers, list): - cls.netscaler_provider = nw_service_providers[0] - - if cls.netscaler_provider.state != 'Enabled': - NetworkServiceProvider.update( - cls.api_client, - id=cls.netscaler_provider.id, - state='Enabled' - ) cls.network_offering = NetworkOffering.create( cls.api_client, - cls.services["network_offering"], + cls.services["nw_off_isolated_netscaler"], conservemode=True ) - 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"] @@ -2568,22 +2025,6 @@ class TestServiceProvider(cloudstackTestCase): cls.network_offering, cls.netscaler, ] - return - - @classmethod - def tearDownClass(cls): - try: - # Enable the service provider - NetworkServiceProvider.update( - cls.api_client, - id=cls.netscaler_provider.id, - state='Enabled' - ) - #Cleanup resources used - cleanup_resources(cls.api_client, cls._cleanup) - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return def setUp(self): self.apiclient = self.testClient.getApiClient() @@ -2595,27 +2036,6 @@ class TestServiceProvider(cloudstackTestCase): domainid=self.domain.id ) self.cleanup = [self.account] - return - - def tearDown(self): - try: - self.debug("Cleaning up the resources") - #Clean up, terminate the created network offerings - cleanup_resources(self.apiclient, self.cleanup) - interval = Configurations.list( - self.apiclient, - name='network.gc.interval' - ) - wait = Configurations.list( - self.apiclient, - name='network.gc.wait' - ) - # Sleep to ensure that all resources are deleted - time.sleep(int(interval[0].value) + int(wait[0].value)) - self.debug("Cleanup complete!") - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return @attr(tags = ["advancedns","test"]) def test_01_create_nw_off_disabled(self): @@ -2655,7 +2075,7 @@ class TestServiceProvider(cloudstackTestCase): with self.assertRaises(Exception): Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account.name, domainid=self.account.domainid, networkofferingid=self.network_offering.id, @@ -2688,7 +2108,7 @@ class TestServiceProvider(cloudstackTestCase): with self.assertRaises(Exception): Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account.name, domainid=self.account.domainid, networkofferingid=self.network_offering.id, @@ -2718,7 +2138,7 @@ class TestServiceProvider(cloudstackTestCase): self.network_offering.id) self.network = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account.name, domainid=self.account.domainid, networkofferingid=self.network_offering.id, @@ -2735,7 +2155,9 @@ class TestServiceProvider(cloudstackTestCase): accountid=self.account.name, domainid=self.account.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network.id)] + networkids=[str(self.network.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network.id) list_vm_response = VirtualMachine.list( @@ -2814,20 +2236,19 @@ class TestServiceProvider(cloudstackTestCase): accountid=self.account.name, domainid=self.account.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network.id)] + networkids=[str(self.network.id)], + templateid=self.template.id, + zoneid=self.zone.id ) - return - - -class TestDeleteNetscaler(cloudstackTestCase): +class TestDeleteNetscaler(_NetScalerDeployVMBase): @classmethod def setUpClass(cls): cls.testClient = super(TestDeleteNetscaler, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() + cls.services = cls.testClient.getParsedTestDataConfig() - 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()) @@ -2837,42 +2258,17 @@ class TestDeleteNetscaler(cloudstackTestCase): cls.services["ostype"] ) - physical_networks = PhysicalNetwork.list( - cls.api_client, - zoneid=cls.zone.id - ) - if isinstance(physical_networks, list): - physical_network = physical_networks[0] + netscaler_config = dict(cls.services["netscaler_VPX"]) + netscaler_config.update({'lbdevicededicated': "False"}) + cls._addNetScaler(netscaler_config) - cls.netscaler = NetScaler.add( - cls.api_client, - cls.services["netscaler"], - physicalnetworkid=physical_network.id - ) - - nw_service_providers = NetworkServiceProvider.list( - cls.api_client, - name='Netscaler', - physicalnetworkid=physical_network.id - ) - if isinstance(nw_service_providers, list): - netscaler_provider = nw_service_providers[0] - - if netscaler_provider.state != 'Enabled': - NetworkServiceProvider.update( - cls.api_client, - id=netscaler_provider.id, - state='Enabled' - ) cls.network_offering = NetworkOffering.create( cls.api_client, - cls.services["network_offering"], + cls.services["nw_off_isolated_netscaler"], conservemode=True ) # Enable Network offering cls.network_offering.update(cls.api_client, state='Enabled') - cls.services["virtual_machine"]["zoneid"] = cls.zone.id - cls.services["virtual_machine"]["template"] = cls.template.id cls.service_offering = ServiceOffering.create( cls.api_client, @@ -2883,54 +2279,6 @@ class TestDeleteNetscaler(cloudstackTestCase): cls.network_offering, cls.netscaler, ] - return - - @classmethod - def tearDownClass(cls): - try: - #Cleanup resources used - cleanup_resources(cls.api_client, cls._cleanup) - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return - - def setUp(self): - self.apiclient = self.testClient.getApiClient() - self.dbclient = self.testClient.getDbConnection() - self.account_1 = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) - self.account_2 = Account.create( - self.apiclient, - self.services["account"], - admin=True, - domainid=self.domain.id - ) - self.cleanup = [self.account_1, self.account_2] - return - - def tearDown(self): - try: - self.debug("Cleaning up the resources") - #Clean up, terminate the created network offerings - cleanup_resources(self.apiclient, self.cleanup) - interval = Configurations.list( - self.apiclient, - name='network.gc.interval' - ) - wait = Configurations.list( - self.apiclient, - name='network.gc.wait' - ) - # Sleep to ensure that all resources are deleted - time.sleep(int(interval[0].value) + int(wait[0].value)) - self.debug("Cleanup complete!") - except Exception as e: - raise Exception("Warning: Exception during cleanup : %s" % e) - return @attr(tags = ["advancedns","test"]) def test_delete_netscaler_with_lb(self): @@ -2973,7 +2321,7 @@ class TestDeleteNetscaler(cloudstackTestCase): self.network_offering.id) self.network_1 = Network.create( self.apiclient, - self.services["network"], + self.services["netscaler_network"], accountid=self.account_1.name, domainid=self.account_1.domainid, networkofferingid=self.network_offering.id, @@ -2990,7 +2338,9 @@ class TestDeleteNetscaler(cloudstackTestCase): accountid=self.account_1.name, domainid=self.account_1.domainid, serviceofferingid=self.service_offering.id, - networkids=[str(self.network_1.id)] + networkids=[str(self.network_1.id)], + templateid=self.template.id, + zoneid=self.zone.id ) self.debug("Deployed VM in network: %s" % self.network_1.id) list_vm_response = VirtualMachine.list( @@ -3019,4 +2369,3 @@ class TestDeleteNetscaler(cloudstackTestCase): with self.assertRaises(Exception): self.netscaler.delete(self.apiclient) self.debug("Attempt to delete Netscaler device failed!") - return diff --git a/tools/marvin/marvin/config/test_data.py b/tools/marvin/marvin/config/test_data.py index ffc3518a1d4..9848616bae7 100644 --- a/tools/marvin/marvin/config/test_data.py +++ b/tools/marvin/marvin/config/test_data.py @@ -304,9 +304,13 @@ test_data = { "lbdevicecapacity": 2, "port": 22 }, - "network_without_acl": { - "name": "TestNetwork", - "displaytext": "TestNetwork", + "netscaler_network": { + "name": "Netscaler", + "displaytext": "Netscaler", + }, + "network_without_acl": { + "name": "TestNetwork", + "displaytext": "TestNetwork", }, "virtual_machine": { "displayname": "Test VM",