Repository: cloudstack Updated Branches: refs/heads/4.5 71c980416 -> 159a372fd
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/159a372f/test/integration/component/test_routers.py ---------------------------------------------------------------------- diff --git a/test/integration/component/test_routers.py b/test/integration/component/test_routers.py index 24f44b3..0a74234 100644 --- a/test/integration/component/test_routers.py +++ b/test/integration/component/test_routers.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,81 +16,101 @@ # under the License. """ P1 tests for routers """ -#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 System modules +from marvin.cloudstackTestCase import cloudstackTestCase +from marvin.cloudstackAPI import (stopVirtualMachine, + stopRouter, + startRouter) +from marvin.lib.utils import (cleanup_resources, + get_process_status, + get_host_credentials) +from marvin.lib.base import (ServiceOffering, + VirtualMachine, + Account, + LoadBalancerRule, + FireWallRule, + NATRule) +from marvin.lib.common import (get_zone, + get_template, + get_domain, + list_virtual_machines, + list_networks, + list_configurations, + list_routers, + list_nat_rules, + list_publicIP, + list_lb_rules, + list_firewall_rules, + list_hosts) + +# Import System modules import time class Services: + """Test router Services """ def __init__(self): self.services = { - "service_offering": { - "name": "Tiny Instance", - "displaytext": "Tiny Instance", - "cpunumber": 1, - "cpuspeed": 100, # in MHz + "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', - }, - "host": { - "username": "root", - "password": "password", - "publicport": 22, - }, - "account": { - "email": "t...@test.com", - "firstname": "Test", - "lastname": "User", - "username": "testuser", - "password": "password", - }, - "natrule": - { - "privateport": 22, - "publicport": 22, - "protocol": "TCP" - }, - "lbrule": - { - "name": "SSH", - "alg": "roundrobin", - # Algorithm used for load balancing - "privateport": 22, - "publicport": 22, - "protocol": 'TCP', - }, - "fw_rule": { - "startport": 1, - "endport": 6000, - "cidr": '55.55.0.0/11', - # Any network (For creating FW rule - }, - "ostype": 'CentOS 5.3 (64-bit)', - # Used for Get_Template : CentOS 5.3 (64 bit) - } + }, + "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', + }, + "host": { + "username": "root", + "password": "password", + "publicport": 22, + }, + "account": { + "email": "t...@test.com", + "firstname": "Test", + "lastname": "User", + "username": "testuser", + "password": "password", + }, + "natrule": + { + "privateport": 22, + "publicport": 22, + "protocol": "TCP" + }, + "lbrule": + { + "name": "SSH", + "alg": "roundrobin", + # Algorithm used for load balancing + "privateport": 22, + "publicport": 22, + "protocol": 'TCP', + }, + "fw_rule": { + "startport": 1, + "endport": 6000, + "cidr": '55.55.0.0/11', + # Any network (For creating FW rule + }, + "ostype": 'CentOS 5.3 (64-bit)', + # Used for Get_Template : CentOS 5.3 (64 bit) + } class TestRouterServices(cloudstackTestCase): @@ -107,50 +127,52 @@ class TestRouterServices(cloudstackTestCase): cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) 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.services["virtual_machine"]["zoneid"] = cls.zone.id - #Create an account, network, VM and IP addresses + # Create an account, network, VM and IP addresses 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.service_offering = ServiceOffering.create( - cls.api_client, - cls.services["service_offering"] - ) + cls.api_client, + cls.services["service_offering"] + ) cls.vm_1 = VirtualMachine.create( - cls.api_client, - cls.services["virtual_machine"], - templateid=cls.template.id, - accountid=cls.account.name, - domainid=cls.account.domainid, - serviceofferingid=cls.service_offering.id - ) + cls.api_client, + cls.services["virtual_machine"], + templateid=cls.template.id, + accountid=cls.account.name, + domainid=cls.account.domainid, + serviceofferingid=cls.service_offering.id + ) cls.vm_2 = VirtualMachine.create( - cls.api_client, - cls.services["virtual_machine"], - templateid=cls.template.id, - accountid=cls.account.name, - domainid=cls.account.domainid, - serviceofferingid=cls.service_offering.id - ) + cls.api_client, + cls.services["virtual_machine"], + templateid=cls.template.id, + accountid=cls.account.name, + domainid=cls.account.domainid, + serviceofferingid=cls.service_offering.id + ) cls._cleanup = [ - cls.account, - cls.service_offering - ] + cls.account, + cls.service_offering + ] return @classmethod def tearDownClass(cls): try: - cls.api_client = super(TestRouterServices, cls).getClsTestClient().getApiClient() - #Clean up, terminate the created templates + cls.api_client = super( + TestRouterServices, + cls).getClsTestClient().getApiClient() + # Clean up, terminate the created templates cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: @@ -184,142 +206,143 @@ class TestRouterServices(cloudstackTestCase): # g. userdata # 2. wait for router to start and guest network to be created # a. listRouters account=user, domainid=1 (router state=Running) - # b. listNetworks account=user domainid=1 (network state=Implemented) + # b. listNetworks account=user domainid=1 + # (network state=Implemented) # c. listVirtualMachines account=user domainid=1 (VM state=Running) # 3. listNetwork routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) self.assertNotEqual( - len(routers), - 0, - "Check list router response" - ) + len(routers), + 0, + "Check list router response" + ) for router in routers: self.debug("Router ID: %s & Router state: %s" % ( - router.id, - router.state - )) + router.id, + router.state + )) self.assertEqual( - router.state, - 'Running', - "Router state is not running but is %s" % router.state - ) + router.state, + 'Running', + "Router state is not running but is %s" % router.state + ) # Network state associated with account should be 'Implemented' networks = list_networks( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - type = 'Isolated' - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + type='Isolated' + ) self.assertEqual( - isinstance(networks, list), - True, - "Check for list networks response return valid data" - ) + isinstance(networks, list), + True, + "Check for list networks response return valid data" + ) self.assertNotEqual( - len(networks), - 0, - "Check list networks response" - ) + len(networks), + 0, + "Check list networks response" + ) for network in networks: self.debug("Network ID: %s & Network state: %s" % ( - network.id, - network.state - )) + network.id, + network.state + )) self.assertIn( - network.state, - ['Implemented', 'Allocated'], - "Check list network response for network state" - ) + network.state, + ['Implemented', 'Allocated'], + "Check list network response for network state" + ) # VM state associated with account should be 'Running' virtual_machines = list_virtual_machines( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid + ) self.assertEqual( - isinstance(virtual_machines, list), - True, - "Check for list virtual machines response return valid data" - ) + isinstance(virtual_machines, list), + True, + "Check for list virtual machines response return valid data" + ) self.assertNotEqual( - len(virtual_machines), - 0, - "Check list virtual machines response" - ) + len(virtual_machines), + 0, + "Check list virtual machines response" + ) for virtual_machine in virtual_machines: self.assertEqual( - virtual_machine.state, - 'Running', - "Check list VM response for Running state" - ) + virtual_machine.state, + 'Running', + "Check list VM response for Running state" + ) self.debug("VM ID: %s & VM state: %s" % ( - virtual_machine.id, - virtual_machine.state - )) + virtual_machine.id, + virtual_machine.state + )) # Check status of DNS, DHCP, FIrewall, LB VPN processes networks = list_networks( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - type = 'Isolated' - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + type='Isolated' + ) self.assertEqual( - isinstance(networks, list), - True, - "Check for list networks response return valid data" - ) + isinstance(networks, list), + True, + "Check for list networks response return valid data" + ) self.assertNotEqual( - len(networks), - 0, - "Check list networks response" - ) + len(networks), + 0, + "Check list networks response" + ) # Load Balancer, Userdata, VPN, Firewall, Gateway, DNS processes should # be running for network in networks: self.assertEqual( - 'Lb' in str(network.service), - True, - "Check Load balancing process in list networks" - ) + 'Lb' in str(network.service), + True, + "Check Load balancing process in list networks" + ) self.assertEqual( - 'UserData' in str(network.service), - True, - "Check UserData service in list networks" - ) + 'UserData' in str(network.service), + True, + "Check UserData service in list networks" + ) self.assertEqual( - 'Vpn' in str(network.service), - True, - "Check Vpn service in list networks" - ) + 'Vpn' in str(network.service), + True, + "Check Vpn service in list networks" + ) self.assertEqual( - 'Firewall' in str(network.service), - True, - "Check Firewall service in list networks" - ) + 'Firewall' in str(network.service), + True, + "Check Firewall service in list networks" + ) self.assertEqual( - 'Dns' in str(network.service), - True, - "Check Dns service in list networks" - ) + 'Dns' in str(network.service), + True, + "Check Dns service in list networks" + ) return - @attr(configuration = "network.gc") + @attr(configuration="network.gc") @attr(tags=["advanced"], required_hardware="false") def test_02_NetworkGarbageCollection(self): """Test network garbage collection @@ -336,36 +359,38 @@ class TestRouterServices(cloudstackTestCase): # 7. listRouters account=user, domainid=1 routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) self.assertNotEqual( - len(routers), - 0, - "Check list router response" - ) + len(routers), + 0, + "Check list router response" + ) # Router associated with account should be in running state timeout = 180 router = routers[0] self.debug("Router ID: %s & Router state: %s" % ( - router.id, - router.state - )) - self.debug("Wait for %s secs max for router to reach Running state" % timeout) + router.id, + router.state + )) + self.debug( + "Wait for %s secs max for router to reach Running state" % + timeout) while timeout: time.sleep(60) routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - id=router.id) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + id=router.id) router = routers[0] if router.state == 'Running': break @@ -373,67 +398,66 @@ class TestRouterServices(cloudstackTestCase): timeout = timeout - 60 if timeout == 0: self.assertEqual( - router.state, - 'Running', - "Router not in Running state") - + router.state, + 'Running', + "Router not in Running state") # Network state associated with account should be 'Implemented' networks = list_networks( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - type = 'Isolated' - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + type='Isolated' + ) self.assertEqual( - isinstance(networks, list), - True, - "Check for list networks response return valid data" - ) + isinstance(networks, list), + True, + "Check for list networks response return valid data" + ) self.assertNotEqual( - len(networks), - 0, - "Check list networks response" - ) + len(networks), + 0, + "Check list networks response" + ) # Check if network in 'Implemented' state for network in networks: self.debug("Network ID: %s & Network state: %s" % ( - network.id, - network.state - )) + network.id, + network.state + )) self.assertIn( - network.state, - ['Implemented', 'Allocated'], - "Check list network response for network state" - ) + network.state, + ['Implemented', 'Allocated'], + "Check list network response for network state" + ) # VM state associated with account should be 'Running' virtual_machines = list_virtual_machines( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(virtual_machines, list), - True, - "Check for list virtual machines response return valid data" - ) + isinstance(virtual_machines, list), + True, + "Check for list virtual machines response return valid data" + ) self.assertNotEqual( - len(virtual_machines), - 0, - "Check list virtual machines response" - ) + len(virtual_machines), + 0, + "Check list virtual machines response" + ) for virtual_machine in virtual_machines: self.debug("VM ID: %s & VM state: %s" % ( - virtual_machine.id, - virtual_machine.state - )) + virtual_machine.id, + virtual_machine.state + )) self.assertEqual( - virtual_machine.state, - 'Running', - "Check list VM response for Running state" - ) + virtual_machine.state, + 'Running', + "Check list VM response for Running state" + ) # Stop virtual machine cmd = stopVirtualMachine.stopVirtualMachineCmd() @@ -441,58 +465,59 @@ class TestRouterServices(cloudstackTestCase): self.apiclient.stopVirtualMachine(cmd) gcinterval = list_configurations( - self.apiclient, - name='network.gc.interval' - ) + self.apiclient, + name='network.gc.interval' + ) self.assertEqual( - isinstance(gcinterval, list), - True, - "Check for list intervals response return valid data" - ) + isinstance(gcinterval, list), + True, + "Check for list intervals response return valid data" + ) self.debug("network.gc.interval: %s" % gcinterval[0].value) gcwait = list_configurations( - self.apiclient, - name='network.gc.wait' - ) + self.apiclient, + name='network.gc.wait' + ) self.assertEqual( - isinstance(gcwait, list), - True, - "Check for list intervals response return valid data" - ) + isinstance(gcwait, list), + True, + "Check for list intervals response return valid data" + ) self.debug("network.gc.wait: %s" % gcwait[0].value) total_wait = int(gcinterval[0].value) + int(gcwait[0].value) # Router is stopped after (network.gc.interval *2) time. Wait for - # (network.gc.interval+network.gc.wait) * 2 for moving router to 'Stopped' + # (network.gc.interval+network.gc.wait) * 2 for moving + # router to 'Stopped' time.sleep(total_wait * 2) routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) self.assertNotEqual( - len(routers), - 0, - "Check list router response" - ) + len(routers), + 0, + "Check list router response" + ) for router in routers: self.debug("Router ID: %s & Router state: %s" % ( - router.id, - router.state - )) + router.id, + router.state + )) self.assertEqual( - router.state, - 'Stopped', - "Check list router response for router state" - ) + router.state, + 'Stopped', + "Check list router response for router state" + ) # Cleanup Vm_2 - Not required for further tests self.cleanup.append(self.vm_2) @@ -506,32 +531,32 @@ class TestRouterServices(cloudstackTestCase): # 1. deployVirtualMachine in the account # 2. listVirtualMachines account=user, domainid=1 # 3. when listVirtualMachines reports the userVM to be in state=Running - # 4. listRouters should report router to have come back to "Running" state + # 4. listRouters should report router to have come back to "Running" + # state # 5. All other VMs in the account should remain in "Stopped" state - - #stop all pre-existing virtual machines if they are in 'Running' state + # stop all pre-existing virtual machines if they are in 'Running' state virtual_machines = list_virtual_machines( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(virtual_machines, list), - True, - "Check for list virtual machines response return valid data" - ) + isinstance(virtual_machines, list), + True, + "Check for list virtual machines response return valid data" + ) self.assertNotEqual( - len(virtual_machines), - 0, - "Check list virtual machines response" - ) + len(virtual_machines), + 0, + "Check list virtual machines response" + ) for virtual_machine in virtual_machines: self.debug("VM ID: %s & VM state: %s" % ( - virtual_machine.id, - virtual_machine.state - )) + virtual_machine.id, + virtual_machine.state + )) if virtual_machine.state == 'Running': # Stop virtual machine cmd = stopVirtualMachine.stopVirtualMachineCmd() @@ -539,101 +564,100 @@ class TestRouterServices(cloudstackTestCase): self.apiclient.stopVirtualMachine(cmd) vm = 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("Deployed a VM with ID: %s" % vm.id) virtual_machines = list_virtual_machines( - self.apiclient, - id=vm.id, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + id=vm.id, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(virtual_machines, list), - True, - "Check for list virtual machines response return valid data" - ) + isinstance(virtual_machines, list), + True, + "Check for list virtual machines response return valid data" + ) self.assertNotEqual( - len(virtual_machines), - 0, - "Check list virtual machines response" - ) + len(virtual_machines), + 0, + "Check list virtual machines response" + ) # VM state should be 'Running' for virtual_machine in virtual_machines: self.assertEqual( - virtual_machine.state, - 'Running', - "Check list VM response for Running state" - ) + virtual_machine.state, + 'Running', + "Check list VM response for Running state" + ) routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) self.assertNotEqual( - len(routers), - 0, - "Check list router response" - ) + len(routers), + 0, + "Check list router response" + ) # Routers associated with account should be 'Running' after deployment # of VM for router in routers: self.debug("Router ID: %s & Router state: %s" % ( - router.id, - router.state - )) + router.id, + router.state + )) self.assertEqual( - router.state, - 'Running', - "Check list router response for router state" - ) - + router.state, + 'Running', + "Check list router response for router state" + ) # All other VMs (VM_1) should be in 'Stopped' virtual_machines = list_virtual_machines( - self.apiclient, - id=self.vm_1.id, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + id=self.vm_1.id, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(virtual_machines, list), - True, - "Check for list VMs response return valid data" - ) + isinstance(virtual_machines, list), + True, + "Check for list VMs response return valid data" + ) self.assertNotEqual( - len(virtual_machines), - 0, - "Check list virtual machines response" - ) + len(virtual_machines), + 0, + "Check list virtual machines response" + ) for virtual_machine in virtual_machines: self.debug("VM ID: %s & VM state: %s" % ( - virtual_machine.id, - virtual_machine.state - )) + virtual_machine.id, + virtual_machine.state + )) self.assertEqual( - virtual_machine.state, - 'Stopped', - "Check list VM response for Stopped state" - ) + virtual_machine.state, + 'Stopped', + "Check list VM response for Stopped state" + ) return @@ -652,41 +676,43 @@ class TestRouterStopCreatePF(cloudstackTestCase): cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.services['mode'] = cls.zone.networktype 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 - #Create an account, network, VM and IP addresses + # Create an account, network, VM and IP addresses 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.service_offering = ServiceOffering.create( - cls.api_client, - cls.services["service_offering"] - ) + cls.api_client, + cls.services["service_offering"] + ) cls.vm_1 = 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 - ) + cls.api_client, + cls.services["virtual_machine"], + templateid=template.id, + accountid=cls.account.name, + domainid=cls.account.domainid, + serviceofferingid=cls.service_offering.id + ) cls._cleanup = [ - cls.account, - cls.service_offering - ] + cls.account, + cls.service_offering + ] return @classmethod def tearDownClass(cls): try: - cls.api_client = super(TestRouterStopCreatePF, cls).getClsTestClient().getApiClient() + cls.api_client = super( + TestRouterStopCreatePF, + cls).getClsTestClient().getApiClient() # Clean up, terminate the created resources cleanup_resources(cls.api_client, cls._cleanup) @@ -725,130 +751,132 @@ class TestRouterStopCreatePF(cloudstackTestCase): # Get router details associated for that account routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) self.assertNotEqual( - len(routers), - 0, - "Check list router response" - ) + len(routers), + 0, + "Check list router response" + ) router = routers[0] self.debug("Stopping router ID: %s" % router.id) - #Stop the router + # Stop the router cmd = stopRouter.stopRouterCmd() cmd.id = router.id self.apiclient.stopRouter(cmd) routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) router = routers[0] self.assertEqual( - router.state, - 'Stopped', - "Check list router response for router state" - ) + router.state, + 'Stopped', + "Check list router response for router state" + ) public_ips = list_publicIP( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - zoneid=self.zone.id - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + zoneid=self.zone.id + ) self.assertEqual( - isinstance(public_ips, list), - True, - "Check for list public IPs response return valid data" - ) + isinstance(public_ips, list), + True, + "Check for list public IPs response return valid data" + ) public_ip = public_ips[0] # Open up firewall port for SSH - fw_rule = FireWallRule.create( - self.apiclient, - ipaddressid=public_ip.id, - protocol=self.services["natrule"]["protocol"], - cidrlist=['0.0.0.0/0'], - startport=self.services["natrule"]["publicport"], - endport=self.services["natrule"]["publicport"] - ) + FireWallRule.create( + self.apiclient, + ipaddressid=public_ip.id, + protocol=self.services["natrule"]["protocol"], + cidrlist=['0.0.0.0/0'], + startport=self.services["natrule"]["publicport"], + endport=self.services["natrule"]["publicport"] + ) self.debug("Creating NAT rule for VM ID: %s" % self.vm_1.id) - #Create NAT rule + # Create NAT rule nat_rule = NATRule.create( - self.apiclient, - self.vm_1, - self.services["natrule"], - public_ip.id - ) + self.apiclient, + self.vm_1, + self.services["natrule"], + public_ip.id + ) self.debug("Starting router ID: %s" % router.id) - #Start the router + # Start the router cmd = startRouter.startRouterCmd() cmd.id = router.id self.apiclient.startRouter(cmd) routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - zoneid=self.zone.id - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + zoneid=self.zone.id + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) router = routers[0] self.assertEqual( - router.state, - 'Running', - "Check list router response for router state" - ) + router.state, + 'Running', + "Check list router response for router state" + ) # NAT Rule should be in Active state after router start nat_rules = list_nat_rules( - self.apiclient, - id=nat_rule.id - ) + self.apiclient, + id=nat_rule.id + ) self.assertEqual( - isinstance(nat_rules, list), - True, - "Check for list NAT rules response return valid data" - ) + isinstance(nat_rules, list), + True, + "Check for list NAT rules response return valid data" + ) self.assertEqual( - nat_rules[0].state, - 'Active', - "Check list port forwarding rules" - ) + nat_rules[0].state, + 'Active', + "Check list port forwarding rules" + ) try: self.debug("SSH into VM with ID: %s" % nat_rule.ipaddress) - self.vm_1.get_ssh_client(ipaddress=nat_rule.ipaddress, port=self.services["natrule"]["publicport"]) + self.vm_1.get_ssh_client( + ipaddress=nat_rule.ipaddress, + port=self.services["natrule"]["publicport"]) except Exception as e: self.fail( - "SSH Access failed for %s: %s" % \ - (nat_rule.ipaddress, e) - ) + "SSH Access failed for %s: %s" % + (nat_rule.ipaddress, e) + ) return @@ -866,42 +894,44 @@ class TestRouterStopCreateLB(cloudstackTestCase): cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.services['mode'] = cls.zone.networktype 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 - #Create an account, network, VM and IP addresses + # Create an account, network, VM and IP addresses 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.service_offering = ServiceOffering.create( - cls.api_client, - cls.services["service_offering"] - ) + cls.api_client, + cls.services["service_offering"] + ) cls.vm_1 = 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 - ) + cls.api_client, + cls.services["virtual_machine"], + templateid=template.id, + accountid=cls.account.name, + domainid=cls.account.domainid, + serviceofferingid=cls.service_offering.id + ) cls._cleanup = [ - cls.account, - cls.service_offering - ] + cls.account, + cls.service_offering + ] return @classmethod def tearDownClass(cls): try: - cls.api_client = super(TestRouterStopCreateLB, cls).getClsTestClient().getApiClient() - #Clean up, terminate the created resources + cls.api_client = super( + TestRouterStopCreateLB, + cls).getClsTestClient().getApiClient() + # Clean up, terminate the created resources cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: @@ -933,126 +963,126 @@ class TestRouterStopCreateLB(cloudstackTestCase): # Get router details associated for that account routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) self.assertNotEqual( - len(routers), - 0, - "Check list router response" - ) + len(routers), + 0, + "Check list router response" + ) router = routers[0] self.debug("Stopping router with ID: %s" % router.id) - #Stop the router + # Stop the router cmd = stopRouter.stopRouterCmd() cmd.id = router.id self.apiclient.stopRouter(cmd) routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) router = routers[0] self.assertEqual( - router.state, - 'Stopped', - "Check list router response for router state" - ) + router.state, + 'Stopped', + "Check list router response for router state" + ) public_ips = list_publicIP( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid + ) self.assertEqual( - isinstance(public_ips, list), - True, - "Check for list public IPs response return valid data" - ) + isinstance(public_ips, list), + True, + "Check for list public IPs response return valid data" + ) public_ip = public_ips[0] # Open up firewall port for SSH - fw_rule = FireWallRule.create( - self.apiclient, - ipaddressid=public_ip.id, - protocol=self.services["lbrule"]["protocol"], - cidrlist=['0.0.0.0/0'], - startport=self.services["lbrule"]["publicport"], - endport=self.services["lbrule"]["publicport"] - ) + FireWallRule.create( + self.apiclient, + ipaddressid=public_ip.id, + protocol=self.services["lbrule"]["protocol"], + cidrlist=['0.0.0.0/0'], + startport=self.services["lbrule"]["publicport"], + endport=self.services["lbrule"]["publicport"] + ) self.debug("Creating LB rule for public IP: %s" % public_ip.id) - #Create Load Balancer rule and assign VMs to rule + # Create Load Balancer rule and assign VMs to rule lb_rule = LoadBalancerRule.create( - self.apiclient, - self.services["lbrule"], - public_ip.id, - accountid=self.account.name - ) + self.apiclient, + self.services["lbrule"], + public_ip.id, + accountid=self.account.name + ) self.debug("Assigning VM %s to LB rule: %s" % ( - self.vm_1.id, - lb_rule.id - )) + self.vm_1.id, + lb_rule.id + )) lb_rule.assign(self.apiclient, [self.vm_1]) - #Start the router + # Start the router cmd = startRouter.startRouterCmd() cmd.id = router.id self.apiclient.startRouter(cmd) routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) router = routers[0] self.assertEqual( - router.state, - 'Running', - "Check list router response for router state" - ) + router.state, + 'Running', + "Check list router response for router state" + ) # After router start, LB RUle should be in Active state lb_rules = list_lb_rules( - self.apiclient, - id=lb_rule.id - ) + self.apiclient, + id=lb_rule.id + ) self.assertEqual( - isinstance(lb_rules, list), - True, - "Check for list LB rules response return valid data" - ) + isinstance(lb_rules, list), + True, + "Check for list LB rules response return valid data" + ) self.assertEqual( - lb_rules[0].state, - 'Active', - "Check list load balancing rules" - ) + lb_rules[0].state, + 'Active', + "Check list load balancing rules" + ) self.assertEqual( - lb_rules[0].publicport, - str(self.services["lbrule"]["publicport"]), - "Check list load balancing rules" - ) + lb_rules[0].publicport, + str(self.services["lbrule"]["publicport"]), + "Check list load balancing rules" + ) try: self.debug("SSH into VM with IP: %s" % public_ip.ipaddress) @@ -1060,9 +1090,9 @@ class TestRouterStopCreateLB(cloudstackTestCase): self.vm_1.get_ssh_client(public_ip.ipaddress) except Exception as e: self.fail( - "SSH Access failed for %s: %s" % \ - (self.vm_1.ipaddress, e) - ) + "SSH Access failed for %s: %s" % + (self.vm_1.ipaddress, e) + ) return @@ -1080,41 +1110,43 @@ class TestRouterStopCreateFW(cloudstackTestCase): cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.services['mode'] = cls.zone.networktype 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 - #Create an account, network, VM and IP addresses + # Create an account, network, VM and IP addresses 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.service_offering = ServiceOffering.create( - cls.api_client, - cls.services["service_offering"] - ) + cls.api_client, + cls.services["service_offering"] + ) cls.vm_1 = 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 - ) + cls.api_client, + cls.services["virtual_machine"], + templateid=template.id, + accountid=cls.account.name, + domainid=cls.account.domainid, + serviceofferingid=cls.service_offering.id + ) cls._cleanup = [ - cls.account, - cls.service_offering - ] + cls.account, + cls.service_offering + ] return @classmethod def tearDownClass(cls): try: - cls.api_client = super(TestRouterStopCreateFW, cls).getClsTestClient().getApiClient() - #Clean up, terminate the created templates + cls.api_client = super( + TestRouterStopCreateFW, + cls).getClsTestClient().getApiClient() + # Clean up, terminate the created templates cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: @@ -1146,147 +1178,148 @@ class TestRouterStopCreateFW(cloudstackTestCase): # Get the router details associated with account routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) self.assertNotEqual( - len(routers), - 0, - "Check list router response" - ) + len(routers), + 0, + "Check list router response" + ) router = routers[0] self.debug("Stopping the router: %s" % router.id) - #Stop the router + # Stop the router cmd = stopRouter.stopRouterCmd() cmd.id = router.id self.apiclient.stopRouter(cmd) routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) router = routers[0] self.assertEqual( - router.state, - 'Stopped', - "Check list router response for router state" - ) + router.state, + 'Stopped', + "Check list router response for router state" + ) public_ips = list_publicIP( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid + ) self.assertEqual( - isinstance(public_ips, list), - True, - "Check for list public IP response return valid data" - ) + isinstance(public_ips, list), + True, + "Check for list public IP response return valid data" + ) public_ip = public_ips[0] - #Create Firewall rule with configurations from settings file + # Create Firewall rule with configurations from settings file fw_rule = FireWallRule.create( - self.apiclient, - ipaddressid=public_ip.id, - protocol='TCP', - cidrlist=[self.services["fw_rule"]["cidr"]], - startport=self.services["fw_rule"]["startport"], - endport=self.services["fw_rule"]["endport"] - ) + self.apiclient, + ipaddressid=public_ip.id, + protocol='TCP', + cidrlist=[self.services["fw_rule"]["cidr"]], + startport=self.services["fw_rule"]["startport"], + endport=self.services["fw_rule"]["endport"] + ) self.debug("Created firewall rule: %s" % fw_rule.id) self.debug("Starting the router: %s" % router.id) - #Start the router + # Start the router cmd = startRouter.startRouterCmd() cmd.id = router.id self.apiclient.startRouter(cmd) routers = list_routers( - self.apiclient, - account=self.account.name, - domainid=self.account.domainid, - ) + self.apiclient, + account=self.account.name, + domainid=self.account.domainid, + ) self.assertEqual( - isinstance(routers, list), - True, - "Check for list routers response return valid data" - ) + isinstance(routers, list), + True, + "Check for list routers response return valid data" + ) router = routers[0] self.assertEqual( - router.state, - 'Running', - "Check list router response for router state" - ) + router.state, + 'Running', + "Check list router response for router state" + ) # After Router start, FW rule should be in Active state fw_rules = list_firewall_rules( - self.apiclient, - id=fw_rule.id, - ) + self.apiclient, + id=fw_rule.id, + ) self.assertEqual( - isinstance(fw_rules, list), - True, - "Check for list FW rules response return valid data" - ) + isinstance(fw_rules, list), + True, + "Check for list FW rules response return valid data" + ) self.assertEqual( - fw_rules[0].state, - 'Active', - "Check list load balancing rules" - ) + fw_rules[0].state, + 'Active', + "Check list load balancing rules" + ) self.assertEqual( - fw_rules[0].startport, - str(self.services["fw_rule"]["startport"]), - "Check start port of firewall rule" - ) + fw_rules[0].startport, + str(self.services["fw_rule"]["startport"]), + "Check start port of firewall rule" + ) self.assertEqual( - fw_rules[0].endport, - str(self.services["fw_rule"]["endport"]), - "Check end port of firewall rule" - ) + fw_rules[0].endport, + str(self.services["fw_rule"]["endport"]), + "Check end port of firewall rule" + ) # For DNS and DHCP check 'dnsmasq' process status if (self.hypervisor.lower() == 'vmware' - or self.hypervisor.lower() == 'hyperv'): + or self.hypervisor.lower() == 'hyperv'): result = get_process_status( - self.apiclient.connection.mgtSvr, - 22, - self.apiclient.connection.user, - self.apiclient.connection.passwd, - router.linklocalip, - 'iptables -t nat -L', - hypervisor=self.hypervisor - ) + self.apiclient.connection.mgtSvr, + 22, + self.apiclient.connection.user, + self.apiclient.connection.passwd, + router.linklocalip, + 'iptables -t nat -L', + hypervisor=self.hypervisor + ) else: hosts = list_hosts( - self.apiclient, - id=router.hostid, - ) + self.apiclient, + id=router.hostid, + ) self.assertEqual( - isinstance(hosts, list), - True, - "Check for list hosts response return valid data" - ) + isinstance(hosts, list), + True, + "Check for list hosts response return valid data" + ) host = hosts[0] - host.user, host.passwd = get_host_credentials(self.config, host.ipaddress) + host.user, host.passwd = get_host_credentials( + self.config, host.ipaddress) try: result = get_process_status( host.ipaddress, @@ -1297,7 +1330,10 @@ class TestRouterStopCreateFW(cloudstackTestCase): 'iptables -t nat -L' ) except KeyError: - self.skipTest("Provide a marvin config file with host credentials to run %s" % self._testMethodName) + self.skipTest( + "Provide a marvin config file with host\ + credentials to run %s" % + self._testMethodName) self.debug("iptables -t nat -L: %s" % result) self.debug("Public IP: %s" % public_ip.ipaddress) @@ -1306,6 +1342,6 @@ class TestRouterStopCreateFW(cloudstackTestCase): res.count(str(public_ip.ipaddress)), 1, "Check public IP address" - ) + ) return
