#!/usr/bin/env python # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # 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 # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. """ Tests for host control state """ from marvin.cloudstackAPI import updateHost from nose.plugins.attrib import attr from marvin.cloudstackTestCase import cloudstackTestCase from marvin.lib.common import (get_domain, get_zone, get_template, list_hosts, list_routers, list_ssvms) from marvin.lib.base import (Account, Domain, Host, ServiceOffering, VirtualMachine) from marvin.sshClient import SshClient import time class TestHostControlState(cloudstackTestCase): @classmethod def setUpClass(cls): cls.testClient = super(TestHostControlState, cls).getClsTestClient() cls.apiclient = cls.testClient.getApiClient() cls.services = cls.testClient.getParsedTestDataConfig() # Get Zone, Domain and templates cls.zone = get_zone(cls.apiclient, cls.testClient.getZoneForTests()) cls.hypervisor = cls.testClient.getHypervisorInfo() cls.hostConfig = cls.config.__dict__["zones"][0].__dict__["pods"][0].__dict__["clusters"][0].__dict__["hosts"][0].__dict__ cls.template = get_template( cls.apiclient, cls.zone.id, cls.hypervisor ) cls.services["virtual_machine"]["zoneid"] = cls.zone.id cls.services["template"] = cls.template.id cls.services["zoneid"] = cls.zone.id cls._cleanup = [] cls.domain = Domain.create( cls.apiclient, cls.services["acl"]["domain1"] ) cls._cleanup.append(cls.domain) cls.account = Account.create( cls.apiclient, cls.services["account"], domainid=cls.domain.id ) cls._cleanup.append(cls.account) cls.service_offering = ServiceOffering.create( cls.apiclient, cls.services["service_offerings"]["tiny"] ) cls._cleanup.append(cls.service_offering) cls.vm = VirtualMachine.create( cls.apiclient, cls.services["virtual_machine"], templateid=cls.template.id, accountid=cls.account.name, domainid=cls.account.domainid, serviceofferingid=cls.service_offering.id ) cls._cleanup.append(cls.vm) @classmethod def tearDownClass(cls): super(TestHostControlState, cls).tearDownClass() def setUp(self): self.apiclient = self.testClient.getApiClient() self.cleanup = [] return def tearDown(self): super(TestHostControlState, self).tearDown() def disable_host(self, id): cmd = updateHost.updateHostCmd() cmd.id = id cmd.allocationstate = "Disable" response = self.apiclient.updateHost(cmd) self.assertEqual(response.resourcestate, "Disabled") def enable_host(self, id): cmd = updateHost.updateHostCmd() cmd.id = id cmd.allocationstate = "Enable" response = self.apiclient.updateHost(cmd) self.assertEqual(response.resourcestate, "Enabled") def get_host_ipaddress(self, hostId): hosts = list_hosts( self.apiclient, type='Routing', id=hostId ) return hosts[0].ipaddress def stop_agent(self, host_ipaddress): SshClient(host_ipaddress, port=22, user=self.hostConfig["username"], passwd=self.hostConfig["password"]).execute\ ("systemctl stop cloudstack-agent || service cloudstack-agent stop") def start_agent(self, host_ipaddress): SshClient(host_ipaddress, port=22, user=self.hostConfig["username"], passwd=self.hostConfig["password"]).execute\ ("systemctl start cloudstack-agent || service cloudstack-agent start") def verify_uservm_host_control_state(self, vm_id, state): list_vms = VirtualMachine.list( self.apiclient, id=vm_id ) vm = list_vms[0] self.assertEqual(vm.hostcontrolstate, state, msg="host control state should be %s, but it is %s" % (state, vm.hostcontrolstate)) def verify_ssvm_host_control_state(self, vm_id, state): list_ssvm_response = list_ssvms( self.apiclient, id=vm_id ) vm = list_ssvm_response[0] self.assertEqual(vm.hostcontrolstate, state, msg="host control state should be %s, but it is %s" % (state, vm.hostcontrolstate)) def verify_router_host_control_state(self, vm_id, state): list_router_response = list_routers( self.apiclient, id=vm_id ) vm = list_router_response[0] self.assertEqual(vm.hostcontrolstate, state, msg="host control state should be %s, but it is %s" % (state, vm.hostcontrolstate)) @attr(tags=["basic", "advanced"], required_hardware="false") def test_uservm_host_control_state(self): """ Verify host control state for user vm """ # 1. verify hostcontrolstate = Enabled # 2. Disable the host, verify hostcontrolstate = Disabled list_vms = VirtualMachine.list( self.apiclient, id=self.vm.id ) host_id = list_vms[0].hostid self.verify_uservm_host_control_state(self.vm.id, "Enabled") self.disable_host(host_id) self.verify_uservm_host_control_state(self.vm.id, "Disabled") if self.hypervisor == "kvm": host_ipaddress = self.get_host_ipaddress(host_id) self.stop_agent(host_ipaddress) time.sleep(5) # wait for the host to be Disconnected self.verify_uservm_host_control_state(self.vm.id, "Offline") self.enable_host(host_id) self.verify_uservm_host_control_state(self.vm.id, "Offline") self.start_agent(host_ipaddress) time.sleep(10) # wait for the host to be Up self.verify_uservm_host_control_state(self.vm.id, "Enabled") else: self.enable_host(host_id) self.verify_uservm_host_control_state(self.vm.id, "Enabled") @attr(tags=["basic", "advanced"], required_hardware="false") def test_ssvm_host_control_state(self): """ Verify host control state for systemvm """ # 1. verify hostcontrolstate = Enabled # 2. Disable the host, verify hostcontrolstate = Disabled list_ssvm_response = list_ssvms( self.apiclient, systemvmtype='secondarystoragevm', state='Running', zoneid=self.zone.id ) self.assertEqual( isinstance(list_ssvm_response, list), True, "Check list response returns a valid list" ) ssvm = list_ssvm_response[0] host_id = ssvm.hostid self.verify_ssvm_host_control_state(ssvm.id, "Enabled") self.disable_host(host_id) self.verify_ssvm_host_control_state(ssvm.id, "Disabled") self.enable_host(host_id) self.verify_ssvm_host_control_state(ssvm.id, "Enabled") @attr(tags=["basic", "advanced"], required_hardware="false") def test_router_host_control_state(self): """ Verify host control state for router """ # 1. verify hostcontrolstate = Enabled # 2. Disable the host, verify hostcontrolstate = Disabled list_router_response = list_routers( self.apiclient, state='Running', listall=True, zoneid=self.zone.id ) self.assertEqual( isinstance(list_router_response, list), True, "Check list response returns a valid list" ) router = list_router_response[0] host_id = router.hostid self.verify_router_host_control_state(router.id, "Enabled") self.disable_host(host_id) self.verify_router_host_control_state(router.id, "Disabled") self.enable_host(host_id) self.verify_router_host_control_state(router.id, "Enabled")