mirror of
https://github.com/apache/cloudstack.git
synced 2025-10-26 08:42:29 +01:00
2127 lines
81 KiB
Python
Executable File
2127 lines
81 KiB
Python
Executable File
# 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 DHCP DNS offload feature
|
|
"""
|
|
from marvin.cloudstackTestCase import cloudstackTestCase, unittest
|
|
from marvin.lib.utils import (random_gen, validateList, cleanup_resources, get_hypervisor_type)
|
|
from marvin.lib.base import (Account,
|
|
NIC,
|
|
ServiceOffering,
|
|
DiskOffering,
|
|
VirtualMachine,
|
|
Network,
|
|
NetworkOffering,
|
|
Configurations,
|
|
Host,
|
|
Template,
|
|
SSHKeyPair,
|
|
Router)
|
|
from marvin.lib.common import (get_zone,
|
|
get_template,
|
|
get_domain
|
|
)
|
|
from nose.plugins.attrib import attr
|
|
from marvin.codes import PASS,FAIL
|
|
from random import randint
|
|
import random,string,time,tempfile,os,re
|
|
|
|
@unittest.skip("Skipping dhcp-dns-offload tests for now, since it requires external dhcp server")
|
|
class TestDeployVMs(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.testClient = super(TestDeployVMs, cls).getClsTestClient()
|
|
cls.api_client = cls.testClient.getApiClient()
|
|
cls.services = cls.testClient.getParsedTestDataConfig()
|
|
cls.hypervisor = (get_hypervisor_type(cls.api_client)).lower()
|
|
cls._cleanup = []
|
|
cls.unsupportedHypervisor = False
|
|
if cls.hypervisor not in ["xenserver", "kvm"]:
|
|
cls.unsupportedHypervisor = True
|
|
return
|
|
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
|
|
cls.domain = get_domain(cls.api_client)
|
|
cls.services['mode'] = cls.zone.networktype
|
|
cls.userdata = "Deploy vm with userdata in shared network"
|
|
cls.template = get_template(
|
|
cls.api_client,
|
|
cls.zone.id,
|
|
cls.services["ostype"]
|
|
)
|
|
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
|
|
cls.services["virtual_machine"]["template"] = cls.template.id
|
|
try:
|
|
cls.service_offering = ServiceOffering.create(
|
|
cls.api_client,
|
|
cls.services["service_offering"]
|
|
)
|
|
cls._cleanup.append(cls.service_offering)
|
|
cls.disk_offering = DiskOffering.create(
|
|
cls.api_client,
|
|
cls.services["disk_offering"]
|
|
)
|
|
cls._cleanup.append(cls.disk_offering)
|
|
cls.network_offering1 = NetworkOffering.create(
|
|
cls.api_client,
|
|
cls.services["nw_off_no_services"],
|
|
)
|
|
cls._cleanup.append(cls.network_offering1)
|
|
# Enable Network offering
|
|
cls.network_offering1.update(cls.api_client, state='Enabled')
|
|
cls.network_offering3 = NetworkOffering.create(
|
|
cls.api_client,
|
|
cls.services["network_offering_shared"],
|
|
)
|
|
cls._cleanup.append(cls.network_offering3)
|
|
# Enable Network offering
|
|
cls.network_offering3.update(cls.api_client, state='Enabled')
|
|
#Create shared network without any services
|
|
cls.shared_network = Network.create(
|
|
cls.api_client,
|
|
cls.services["network2"],
|
|
networkofferingid=cls.network_offering1.id,
|
|
zoneid=cls.zone.id
|
|
)
|
|
cls._cleanup.append(cls.shared_network)
|
|
#Verify network services in a created network
|
|
assert TestDeployVMs.list_nw_services(cls.shared_network.id, cls.network_offering1.id),\
|
|
"Network Services does not match with the offering services"
|
|
#Register passowrd and ssh-key reset enabled template
|
|
template = "ssh-password-template-xen" if cls.hypervisor == "xenserver" else "ssh-password-template-kvm"
|
|
cls.pwd_template = Template.register(cls.api_client,
|
|
cls.services[template],
|
|
zoneid=cls.zone.id)
|
|
cls.pwd_template.download(cls.api_client)
|
|
cls._cleanup.append(cls.pwd_template)
|
|
except Exception as e:
|
|
cls.tearDownClass()
|
|
raise e
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
try:
|
|
cls._cleanup = cls._cleanup[::-1]
|
|
cleanup_resources(cls.api_client, cls._cleanup)
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
self.tmp_files = []
|
|
if self.unsupportedHypervisor:
|
|
self.skipTest("Skipping tests because suitable hypervisors not present")
|
|
|
|
def tearDown(self):
|
|
try:
|
|
self.cleanup = self.cleanup[::-1]
|
|
cleanup_resources(self.apiclient, self.cleanup)
|
|
for tmp_file in self.tmp_files:
|
|
os.remove(tmp_file)
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
return
|
|
|
|
def getVMIPAddress(self,id):
|
|
"""
|
|
This method retrieves the IP address from a deployed vm.
|
|
"""
|
|
params = ["externaldhcp.vmip.retrieval.interval","externaldhcp.vmip.max.retry"]
|
|
offload_params = {}
|
|
for param in params:
|
|
configs = Configurations.list(
|
|
self.apiclient,
|
|
name=param,
|
|
)
|
|
self.assertEqual(validateList(configs)[0],PASS,"List config returned invalid response")
|
|
config = configs[0]
|
|
offload_params[param] = config
|
|
max_wait = int(offload_params[params[0]].value)*int(offload_params[params[1]].value)
|
|
wait = 0
|
|
status = 0
|
|
while(wait <= max_wait):
|
|
vm_res = VirtualMachine.list(self.apiclient,id=id)
|
|
self.assertEqual(validateList(vm_res)[0], PASS, "vm list returned invalid response")
|
|
for nic in vm_res[0].nic:
|
|
if nic.type == "Shared":
|
|
if nic.ipaddress is None:
|
|
time.sleep(int(offload_params[params[0]].value))
|
|
wait += int(offload_params[params[0]].value)
|
|
status = 0
|
|
break
|
|
else:
|
|
status = 1
|
|
else:
|
|
continue
|
|
if status == 1:
|
|
self.debug("CS has retrieved the vm ip address from the hypervisor")
|
|
break
|
|
return [status, vm_res[0].nic[0].ipaddress]
|
|
|
|
def list_nics(self, vm_id):
|
|
list_vm_res = VirtualMachine.list(self.apiclient, id=vm_id)
|
|
self.assertEqual(validateList(list_vm_res)[0], PASS, "List vms returned invalid response")
|
|
nics = list_vm_res[0].nic
|
|
for nic in nics:
|
|
if nic.type == "Shared":
|
|
nic_res = NIC.list(
|
|
self.apiclient,
|
|
virtualmachineid=vm_id,
|
|
nicid=nic.id
|
|
)
|
|
nic_ip = nic_res[0].ipaddress
|
|
self.assertIsNotNone(nic_ip, "listNics API response does not have the ip address")
|
|
else:
|
|
continue
|
|
return
|
|
|
|
def isConfigDriveIsoAttached(self, vm_ip):
|
|
"""
|
|
This method is to verify whether configdrive iso is attached to vm or not
|
|
Takes vm ip as the input
|
|
Returns mount path if config drive is attached else False
|
|
"""
|
|
mountdir = "/root/iso"
|
|
cmd = "blkid -t LABEL='config' /dev/hd? /dev/sd? /dev/xvd? -o device"
|
|
try:
|
|
self.debug("SSH into VM: %s" % vm_ip)
|
|
ssh = self.vm.get_ssh_client(ipaddress=vm_ip, reconnect=True)
|
|
tmp_cmd = ['bash -c "if [ ! -d /root/iso ] ; then mkdir /root/iso ; fi"', "umount /root/iso"]
|
|
for tcmd in tmp_cmd:
|
|
ssh.execute(tcmd)
|
|
configDrive = ssh.execute(cmd)
|
|
res = ssh.execute("mount {} {}".format(str(configDrive[0]), mountdir))
|
|
if str(res).lower().find("mounting read-only") > -1:
|
|
self.debug("configDrive iso has been mounted at location %s" % mountdir)
|
|
return mountdir
|
|
else:
|
|
return False
|
|
except Exception as e:
|
|
self.debug("SSH Access failed for vm {}: Exception {}".format(vm_ip, e))
|
|
return False
|
|
|
|
def verifyUserData(self, vm_ip, iso_path):
|
|
"""
|
|
Connect to vm and verify Userdata
|
|
"""
|
|
userdata_path = iso_path+"/cloudstack/userdata/user_data.txt"
|
|
try:
|
|
self.debug("SSH into VM: %s" % vm_ip)
|
|
ssh = self.vm.get_ssh_client(ipaddress=vm_ip, reconnect=True)
|
|
cmd = "cat %s" % userdata_path
|
|
res = ssh.execute(cmd)
|
|
if str(res).lower().find(self.userdata):
|
|
self.debug("Userdata is found in the attached configDriveIso")
|
|
return True
|
|
else:
|
|
self.debug("Userdata is not found in the attached configDriveIso")
|
|
return False
|
|
except Exception as e:
|
|
self.debug("SSH Access failed for vm {}: Exception {}".format(vm_ip, e))
|
|
return False
|
|
|
|
return
|
|
|
|
def verifyPassword(self, vm_ip, iso_path):
|
|
password_file = iso_path+"/cloudstack/password/vm_password.txt"
|
|
try:
|
|
self.debug("SSH into VM: %s" % vm_ip)
|
|
ssh = self.vm.get_ssh_client(ipaddress=vm_ip, reconnect=True)
|
|
cmd = "cat %s" % password_file
|
|
res = ssh.execute(cmd)
|
|
if str(res).find("No such file or directory"):
|
|
self.debug("Password file is not found")
|
|
return False
|
|
elif str(res).lower().find("saved_password"):
|
|
self.debug("Reset password is not found in the attached configDriveIso")
|
|
return False
|
|
else:
|
|
self.debug("Reset Password is found in the attached configDriveIso")
|
|
return True
|
|
except Exception as e:
|
|
self.debug("SSH Access failed for vm {}: Exception {}".format(vm_ip, e))
|
|
return False
|
|
return
|
|
|
|
def verifySshKey(self, vm_ip, iso_path):
|
|
|
|
publicKey_file = iso_path+"/cloudstack/metadata/public_keys.txt"
|
|
try:
|
|
self.debug("SSH into VM: %s" % vm_ip)
|
|
ssh = self.vm.get_ssh_client(ipaddress=vm_ip)
|
|
cmd = "cat %s" % publicKey_file
|
|
res = ssh.execute(cmd)
|
|
if str(res).lower().find("saved_password"):
|
|
self.debug("Public Key is found in the attached configDriveIso")
|
|
return True
|
|
else:
|
|
self.debug("Public Key is not found in the attached configDriveIso")
|
|
return False
|
|
except Exception as e:
|
|
self.debug("SSH Access failed for vm {}: Exception {}".format(vm_ip, e))
|
|
return False
|
|
return
|
|
|
|
def getVMMetaData(self, vm_ip, iso_path):
|
|
|
|
metadata_dir = iso_path+"/cloudstack/metadata/"
|
|
try:
|
|
response = {}
|
|
self.debug("SSH into VM: %s" % vm_ip)
|
|
ssh = self.vm.get_ssh_client(ipaddress=vm_ip, reconnect=True)
|
|
vm_files = ["availability_zone.txt", "instance_id.txt", "service_offering.txt", "vm_id.txt"]
|
|
for file in vm_files:
|
|
cmd = "cat %s" % metadata_dir+file
|
|
res = ssh.execute(cmd)
|
|
response[file] = res
|
|
except Exception as e:
|
|
self.debug("SSH Access failed for vm {}: Exception {}".format(vm_ip, e))
|
|
return response
|
|
|
|
def verifyMetaData(self, metadata):
|
|
|
|
metadata_files = ["availability_zone.txt", "instance_id.txt", "service_offering.txt", "vm_id.txt"]
|
|
for mfile in metadata_files:
|
|
if mfile not in metadata:
|
|
self.fail("{} file is not found in vm metadata".format(mfile))
|
|
self.assertEqual(
|
|
str(metadata["availability_zone.txt"][0]),
|
|
self.zone.name,
|
|
"Zone name inside metadata does not match with the zone"
|
|
)
|
|
self.assertEqual(
|
|
str(metadata["instance_id.txt"][0]),
|
|
self.vm.instancename,
|
|
"vm name inside metadata does not match with the instance name"
|
|
)
|
|
self.assertEqual(
|
|
str(metadata["service_offering.txt"][0]),
|
|
self.vm.serviceofferingname,
|
|
"Service offering inside metadata does not match with the instance offering"
|
|
)
|
|
qresultset = self.dbclient.execute(
|
|
"select id from vm_instance where instance_name='%s';" % \
|
|
self.vm.instancename)
|
|
self.assertEqual(validateList(qresultset)[0], PASS, "sql query returned invalid response")
|
|
self.assertEqual(
|
|
metadata["vm_id.txt"][0],
|
|
unicode(qresultset[0][0]),
|
|
"vm id in metadata does not match with the vm id from cloud db"
|
|
)
|
|
return
|
|
|
|
def generate_ssh_keys(self):
|
|
"""
|
|
This method generates ssh key pair and writes the private key into a temp file and returns the file name
|
|
"""
|
|
self.keypair = SSHKeyPair.create(
|
|
self.apiclient,
|
|
name=random_gen() + ".pem",
|
|
account="admin",
|
|
domainid=1
|
|
)
|
|
self.debug("Created keypair with name: %s" % self.keypair.name)
|
|
self.debug("Writing the private key to local file")
|
|
keyPairFilePath = tempfile.gettempdir() + os.sep + self.keypair.name
|
|
self.tmp_files.append(keyPairFilePath)
|
|
self.debug("File path: %s" % keyPairFilePath)
|
|
f = open(keyPairFilePath, "w+")
|
|
f.write(self.keypair.privatekey)
|
|
f.close()
|
|
os.system("chmod 400 " + keyPairFilePath)
|
|
return keyPairFilePath
|
|
|
|
def verifyIPFetchEvent(self, ipaddr):
|
|
|
|
id = self.dbclient.execute(
|
|
"select instance_id,id from nics where ip4_address='%s' and removed is null;" % ipaddr
|
|
)
|
|
self.assertEqual(
|
|
validateList(id)[0],
|
|
PASS,
|
|
"sql query to select nic id returned invalid response"
|
|
)
|
|
qresultset = self.dbclient.execute(
|
|
"select description from event where type='EXTERNAL.DHCP.VM.IP.FETCH' order by id desc limit 1;"
|
|
)
|
|
self.assertEqual(
|
|
validateList(qresultset)[0],
|
|
PASS,
|
|
"sql query for IP Fetch event returned invalid response"
|
|
)
|
|
return True if re.search("VM %s nic id %s ip address %s got fetched successfully" % \
|
|
(id[0][0], id[0][1], ipaddr), str(qresultset[0])) else False
|
|
|
|
def umountConfigDrive(self, vm_ip, iso_path):
|
|
"""umount config drive iso attached inside guest vm"""
|
|
try:
|
|
self.debug("SSH into VM: %s" % vm_ip)
|
|
ssh = self.vm.get_ssh_client(ipaddress=vm_ip, reconnect=True)
|
|
ssh.execute("umount %s" % iso_path)
|
|
except Exception as e:
|
|
self.debug("SSH access failed for vm")
|
|
|
|
@classmethod
|
|
def list_nw_services(cls, nw_id, nw_off_id):
|
|
"""
|
|
Verify network services in a network. List of services supported should be as per the
|
|
network offering with which it has been created.
|
|
return True if created network and network offering has same services else False
|
|
"""
|
|
nw_response = Network.list(cls.api_client, id = nw_id)
|
|
assert validateList(nw_response)[0] == PASS, "list networks returned invalid response"
|
|
nw_services = [service.name for service in nw_response[0].service]
|
|
nw_off_res = NetworkOffering.list(cls.api_client, id = nw_off_id)
|
|
assert validateList(nw_off_res)[0] == PASS, "list network offerings returned invalid response"
|
|
nw_off_services = [service.name for service in nw_off_res[0].service]
|
|
if len(nw_services) == len(nw_off_services) and len(nw_services) == 0:
|
|
return True
|
|
elif len(nw_services) != len(nw_off_services):
|
|
return False
|
|
for srv in nw_services:
|
|
if srv not in nw_off_services:
|
|
return False
|
|
return True
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_01_createNetworks_TC3(self):
|
|
|
|
"""
|
|
Create shared networks with all the network offerings created in the setupClass method
|
|
"""
|
|
nw_offs = [self.network_offering3]
|
|
vlan = self.services["network2"]["vlan"]
|
|
name = self.services["network2"]["name"]
|
|
for nw_off in nw_offs:
|
|
self.services["network2"]["name"]="SharedNetwork%s" % randint(1, 10)
|
|
try:
|
|
self.services["network2"]["vlan"] += 1
|
|
self.shared_network = Network.create(
|
|
self.apiclient,
|
|
self.services["network2"],
|
|
networkofferingid=nw_off.id,
|
|
zoneid=self.zone.id
|
|
)
|
|
self.cleanup.append(self.shared_network)
|
|
self.assertTrue(
|
|
TestDeployVMs.list_nw_services(self.shared_network.id, nw_off.id),
|
|
"Network services not matched with network offering services"
|
|
)
|
|
except Exception as e:
|
|
self.fail("Shared network creation with network offering %s failed with error: %s" % (nw_off, e))
|
|
self.services["network2"]["vlan"] = vlan
|
|
self.services["network2"]["name"] = name
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_02_deployVM_nw_without_services_TC6(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm in that network
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval
|
|
"""
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"VM creation failed in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
#Verify IP address retrieval using listVirtualMachines API
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
#Verify the retrieved ip address in listNICs API response
|
|
self.list_nics(self.vm.id)
|
|
#Verify config drive is attached to vm or not
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_03_deployVM_with_userdata_TC7(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with userdata in the above network
|
|
3.Verify vm deployment
|
|
4.Verify IP Address retrieval
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"VM creation with userdata failed in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
#Verify IP address retrieval using listVirtualMachines API
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
#Verify the retrieved ip address in listNICs API response
|
|
self.list_nics(self.vm.id)
|
|
#Verify config drive is attached to vm or not
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_04_deploy_win_VM_with_userdata_TC8(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy windows vm in the above network
|
|
3.Verify IP address retrieval
|
|
"""
|
|
self.skipTest("Skip windows tests for now")
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.shared_network = Network.create(
|
|
self.apiclient,
|
|
self.services["network2"],
|
|
networkofferingid=self.network_offering1.id,
|
|
zoneid=self.zone.id
|
|
)
|
|
self.cleanup.append(self.shared_network)
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.win_template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"FAIL:VM Creation failed with windows template in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm_id)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_06_deployVM_with_userdata_TC11(self):
|
|
"""
|
|
1.Create shared network with DHCP,DNS and userdata services provided by VR
|
|
2.Deploy vm in that network
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval
|
|
"""
|
|
vlan = self.services["network2"]["vlan"]
|
|
self.services["network2"]["vlan"] += 1
|
|
self.shared_network = Network.create(
|
|
self.apiclient,
|
|
self.services["network2"],
|
|
networkofferingid=self.network_offering3.id,
|
|
zoneid=self.zone.id
|
|
)
|
|
self.cleanup.append(self.shared_network)
|
|
self.services["network2"]["vlan"] = vlan
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm with in a shared network with DHCP and userdata services provided by VR"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
#To make sure that vm got the ip address as part of the cs allocation process check user_ip_address table
|
|
#for allocated state for the assigned ip address
|
|
qresultset = self.dbclient.execute(
|
|
"select allocated from user_ip_address where public_ip_address='%s' and removed is null;" % \
|
|
self.vm.nic[0].ipaddress)
|
|
self.assertEqual(validateList(qresultset)[0],PASS,"sql query returned invalid response")
|
|
self.assertNotEqual(
|
|
qresultset[0],
|
|
'NULL',
|
|
"VM didn't get ip address as part of CS IP allocation process\
|
|
It might have got from external dhcp server in that network")
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_07_deployVM_with_userdata_VR_TC12(self):
|
|
"""
|
|
1.Create shared network with only userdata service and VR as the provider
|
|
2.Verify that network offering creation should fail
|
|
"""
|
|
try:
|
|
self.network_offering4 = NetworkOffering.create(
|
|
self.api_client,
|
|
self.services["nw_off_userdata_VR"],
|
|
)
|
|
self.fail("NO creation with vr as the userdata provider should not be allowed\
|
|
since VR in not the dhcp provider")
|
|
except Exception as e:
|
|
self.debug("NO creation failed as expected with error message %s" % e)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_09_StopStartVMAndGetVMIpAddress_TC16(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with default cent os template and pass user data
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval and userdata
|
|
5.Stop start vm and repeat step4
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
#Verify configdrive iso attach before vm stop/start
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm after vm stop start"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching with the data passed in vm deployment"
|
|
)
|
|
#Stop/Start VM and verify the IP address retrieval
|
|
self.vm.stop(self.apiclient)
|
|
self.vm.start(self.apiclient)
|
|
time.sleep(60)
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm stop start"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm after vm stop start"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching after vm stop start"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_10_RebootVMAndGetVMIpAddress_TC17(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with default cent os template and pass user data
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval and userdata
|
|
5.Reboot vm and repeat step4
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
#Verify configdrive iso attach before vm reboot
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm before reboot"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching with the data passed while deploying"
|
|
)
|
|
#Reboot VM and verify the IP address retrieval
|
|
self.vm.reboot(self.apiclient)
|
|
time.sleep(60)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm reboot"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm after vm reboot"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching after vm reboot"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_11_DestroyVMAndGetVMIpAddress_TC18(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with default cent os template and pass user data
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval and userdata
|
|
5.Destroy,restore vm and repeat step4
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
#Verify configdrive iso attach before vm destroy
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm before destroy"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching with the data passed while deploying"
|
|
)
|
|
#Destroy VM and verify the IP address retrieval
|
|
self.vm.delete(self.apiclient,expunge=False)
|
|
self.vm.recover(self.apiclient)
|
|
self.vm.start(self.apiclient)
|
|
time.sleep(60)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm reboot"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm after vm reboot"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching after vm reboot"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_12_ReInstallVMAndGetVMIpAddress_TC19(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with default cent os template and pass user data
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval and userdata
|
|
5.Reinstall vm to same template and repeat step4
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
#Verify configdrive iso attach before vm reinstall
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm before reinstall"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr,configDrive_iso),
|
|
True,
|
|
"Userdata is not matching with the data passed while deploying"
|
|
)
|
|
#Reinstall VM and verify the IP address retrieval
|
|
self.vm.restore(self.apiclient)
|
|
time.sleep(60)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm reinstall"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm after vm reinstall"
|
|
)
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching after vm reinstall"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_13_MigrateVMAndGetVMIpAddress_TC20(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with default cent os template and pass user data
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval and userdata
|
|
5.Live Migrate vm to another host in the cluster and repeat step4
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm before live migrate"
|
|
)
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching with the data passed while deploying"
|
|
)
|
|
self.umountConfigDrive(ip_addr, configDrive_iso)
|
|
hosts = Host.listForMigration(self.apiclient,virtualmachineid=self.vm.id)
|
|
self.assertEqual(validateList(hosts)[0], PASS, "Find hosts for migration returned invalid response")
|
|
if len(hosts) < 1:
|
|
self.skipTest("No suitable hosts found for vm migration..so skipping test")
|
|
hostid = hosts[0].id
|
|
self.vm.migrate(self.apiclient,hostid=hostid)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm live migrate"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm after vm live migrate"
|
|
)
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching after vm live migrate"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_14_VM_with_password_enabled_template_TC24(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with password enabled template
|
|
3.Verify accessing vm with the CS generated password
|
|
"""
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.pwd_template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services with password enabled template"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
try:
|
|
self.debug("SSHing into VM: %s" % ip_addr)
|
|
self.vm.get_ssh_client(ipaddress=ip_addr, reconnect=True)
|
|
except Exception as e:
|
|
self.fail("SSH into VM: %s failed with new password: %s" %
|
|
(self.vm.ssh_ip, e))
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_15_reset_password_TC27(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with password enabled template
|
|
3.Reset password
|
|
4.Verify accessing vm with the new password
|
|
"""
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.pwd_template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
self.vm.stop(self.apiclient)
|
|
self.debug("Resetting VM password for VM: %s" % self.vm.name)
|
|
self.vm_password = self.vm.resetPassword(self.apiclient)
|
|
self.debug("Password reset to: %s" % self.vm_password)
|
|
self.debug("Starting VM to verify new password..")
|
|
self.vm.start(self.apiclient)
|
|
self.debug("VM - %s stated!" % self.vm.name)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
try:
|
|
self.debug("SSHing into VM: %s" % ip_addr)
|
|
self.vm.password = self.vm_password
|
|
self.vm.get_ssh_client(ipaddress=ip_addr, reconnect=True)
|
|
except Exception as e:
|
|
self.fail("SSH into VM: %s failed with new password: %s" %
|
|
(self.vm.ssh_ip, e))
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_16_deployVM_with_sshkeys_TC26(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with ssh key reset script enabled template
|
|
3.Verify vm access with new keys
|
|
"""
|
|
#generate key pair and get private key file name by calling below function
|
|
ssh_pvt_key = self.generate_ssh_keys()
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.pwd_template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
keypair=self.keypair.name
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm with ssh key pair"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
try:
|
|
self.debug("SSHing into VM: %s" % ip_addr)
|
|
self.vm.get_ssh_client(ipaddress=ip_addr, reconnect=True, keyPairFileLocation=str(ssh_pvt_key))
|
|
except Exception as e:
|
|
self.fail("SSH into VM: %s failed with ssh key pair: %s" %
|
|
(self.vm.ssh_ip, e))
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_17_reset_sshkeys_TC29(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with ssh key reset script enabled template
|
|
3.Reset keys
|
|
4.Verify vm access with new keys
|
|
"""
|
|
#generate key pair and get private key file name by calling below function
|
|
self.generate_ssh_keys()
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.pwd_template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
keypair=self.keypair.name
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm with ssh key pair"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
try:
|
|
self.vm.stop(self.apiclient)
|
|
except Exception as e:
|
|
self.fail("Failed to stop virtual machine %s with error %s" % (self.vm.id, e))
|
|
self.debug("Creating a new SSH keypair")
|
|
new_pvt_key = self.generate_ssh_keys()
|
|
try:
|
|
self.vm.resetSshKey(
|
|
self.apiclient,
|
|
keypair=self.keypair.name,
|
|
name=self.keypair.name,
|
|
account="admin",
|
|
domainid=1
|
|
)
|
|
except Exception as e:
|
|
self.fail("Failed to reset SSH key: %s, %s" % (self.vm.name, e))
|
|
self.debug("Starting the virtual machine after resetting the keypair")
|
|
try:
|
|
self.vm.start(self.apiclient)
|
|
except Exception as e:
|
|
self.fail("Failed to start virtual machine: %s, %s" % (self.vm.name, e))
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
try:
|
|
self.debug("SSHing into VM: %s" % ip_addr)
|
|
self.vm.get_ssh_client(ipaddress=ip_addr, reconnect=True, keyPairFileLocation=str(new_pvt_key))
|
|
except Exception as e:
|
|
self.fail("SSH into VM: %s failed with ssh key pair with error: %s" %
|
|
(self.vm.ssh_ip, e))
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_18_reset_password_verify_after_vm_restart_TC36(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with password enabled template
|
|
3.Reset password
|
|
4.Verify accessing vm with the new password
|
|
5.Stop/Start vm and repeat step 4
|
|
"""
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.pwd_template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
self.vm.stop(self.apiclient)
|
|
self.debug("Resetting VM password for VM: %s" % self.vm.name)
|
|
self.vm_password = self.vm.resetPassword(self.apiclient)
|
|
self.debug("Password reset to: %s" % self.vm_password)
|
|
self.debug("Starting VM to verify new password..")
|
|
self.vm.start(self.apiclient)
|
|
self.debug("VM - %s stated!" % self.vm.name)
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
try:
|
|
self.debug("SSHing into VM: %s" % ip_addr)
|
|
self.vm.password = self.vm_password
|
|
self.vm.get_ssh_client(ipaddress=ip_addr, reconnect=True)
|
|
except Exception as e:
|
|
self.fail("SSH into VM: %s failed with new password: %s" %
|
|
(self.vm.ssh_ip, e))
|
|
#Stop start vm and verify connecting to vm with the same password
|
|
self.vm.stop(self.apiclient)
|
|
self.vm.start(self.apiclient)
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
try:
|
|
self.debug("SSH into vm after stop start")
|
|
self.vm.password = self.vm_password
|
|
self.vm.get_ssh_client(ipaddress=ip_addr, reconnect=True)
|
|
except Exception as e:
|
|
self.fail("SSH into vm failed with the same password after vm stop start")
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_19_reset_sshkeys_verify_after_vm_restartTC37(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with ssh key reset script enabled template
|
|
3.Reset keys
|
|
4.Verify vm access with new keys
|
|
5.Stop/start vm and repeat step4
|
|
"""
|
|
#generate key pair and get private key file name by calling below function
|
|
self.generate_ssh_keys()
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.pwd_template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
keypair=self.keypair.name
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm with ssh key pair"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
try:
|
|
self.vm.stop(self.apiclient)
|
|
except Exception as e:
|
|
self.fail("Failed to stop virtual machine %s with error %s" % (self.vm.id, e))
|
|
self.debug("Creating a new SSH keypair")
|
|
new_pvt_key = self.generate_ssh_keys()
|
|
try:
|
|
self.vm.resetSshKey(
|
|
self.apiclient,
|
|
keypair=self.keypair.name,
|
|
name=self.keypair.name,
|
|
account="admin",
|
|
domainid=1
|
|
)
|
|
except Exception as e:
|
|
self.fail("Failed to reset SSH key: %s, %s" % (self.vm.name, e))
|
|
self.debug("Starting the virtual machine after resetting the keypair")
|
|
try:
|
|
self.vm.start(self.apiclient)
|
|
except Exception as e:
|
|
self.fail("Failed to start virtual machine: %s, %s" % (self.vm.name, e))
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
try:
|
|
self.debug("SSHing into VM: %s" % ip_addr)
|
|
self.vm.get_ssh_client(ipaddress=ip_addr, reconnect=True, keyPairFileLocation=str(new_pvt_key))
|
|
except Exception as e:
|
|
self.fail("SSH into VM: %s failed with ssh key pair with error: %s" %
|
|
(self.vm.ssh_ip, e))
|
|
self.vm.stop(self.apiclient)
|
|
self.vm.start(self.apiclient)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
try:
|
|
self.debug("SSH into vm after vm stop/start")
|
|
self.vm.get_ssh_client(ipaddress=ip_addr, reconnect=True, keyPairFileLocation=str(new_pvt_key))
|
|
except Exception as e:
|
|
self.fail("SSH into vm failed with new ssh keypair after vm stop start")
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_20_verify_VM_metadata_TC30(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with userdata in the above network
|
|
3.Verify vm deployment
|
|
4.Verify IP Address retrieval
|
|
5.Verify metadata inside configdrive iso
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm with in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
#Verify config drive is attached to vm or not
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_21_StopStartVM_verify_ConfigDrive_TC31(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with default cent os template and pass user data
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval and userdata
|
|
5.Stop start vm
|
|
6.Verify userdata and metadata
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
#Verify configdrive iso attach before vm stop/start
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm before vm stop start"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr,configDrive_iso),
|
|
True,
|
|
"Userdata is not matching with the data passed in vm deployment"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
#Stop/Start VM and verify IP address retrieval,userdata and metadata
|
|
self.vm.stop(self.apiclient)
|
|
self.vm.start(self.apiclient)
|
|
time.sleep(60)
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm stop start"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm after vm stop start"
|
|
)
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching after vm stop start"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_22_RebootVM_verify_ConfigDrive_TC32(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with default cent os template and pass user data
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval,userdata and metadata
|
|
5.Reboot vm
|
|
6.Repeat step4
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
#Verify configdrive iso attach before vm stop/start
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm before vm reboot"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching with the data passed in vm deployment"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
#Reboot VM and verify IP address retrieval,userdata and metadata
|
|
self.vm.reboot(self.apiclient)
|
|
time.sleep(60)
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm reboot"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm after vm reboot"
|
|
)
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching after vm reboot"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_23_DestroyVM_verify_ConfigDrive_TC33(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with default cent os template and pass user data
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval,userdata and metadata
|
|
5.Destroy and restore vm
|
|
6.Repeat step4
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
#Verify configdrive iso attach before vm stop/start
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm before vm destroy"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching with the data passed in vm deployment"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
#Destroy VM and verify IP address retrieval,userdata and metadata
|
|
self.vm.delete(self.apiclient, expunge=False)
|
|
self.vm.recover(self.apiclient)
|
|
self.vm.start(self.apiclient)
|
|
time.sleep(60)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm destroy"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm after vm destroy"
|
|
)
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching after vm destroy"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_24_ReInstallVM_verify_ConfigDrive_TC34(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with default cent os template and pass user data
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval,userdata and metadata
|
|
5.Reinstall vm to the same template
|
|
6.Repeat step4
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
#Verify configdrive iso attach before vm stop/start
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm before vm reinstall"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching with the data passed in vm reinstall"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
#Reinstall VM and verify the IP address retrieval
|
|
self.vm.restore(self.apiclient)
|
|
time.sleep(60)
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm reinstall"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm after vm reinstall"
|
|
)
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr,configDrive_iso),
|
|
True,
|
|
"Userdata is not matching after vm reinstall"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_25_MigrateVMVM_verify_ConfigDrive_TC35(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm with default cent os template and pass user data
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval,userdata and metadata
|
|
5.Migrate VM and Repeat step4
|
|
"""
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
diskofferingid=self.disk_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"Failed to create vm in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
#Verify configdrive iso attach before vm stop/start
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm before vm reinstall"
|
|
)
|
|
#Add userdata verification test
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr, configDrive_iso),
|
|
True,
|
|
"Userdata is not matching with the data passed in vm reinstall"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
#Migrate VM and verify ip address retrieval,userdata and metadata
|
|
hosts = Host.listForMigration(self.apiclient,virtualmachineid=self.vm.id)
|
|
self.assertEqual(validateList(hosts)[0], PASS, "Find hosts for migration returned invalid response")
|
|
if len(hosts) < 1:
|
|
self.skipTest("No suitable hosts found for vm migration..so skipping test")
|
|
hostid = hosts[0].id
|
|
self.umountConfigDrive(ip_addr, configDrive_iso)
|
|
self.vm.migrate(self.apiclient, hostid=hostid)
|
|
time.sleep(60)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm reinstall"
|
|
)
|
|
self.list_nics(self.vm.id)
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm after vm reinstall"
|
|
)
|
|
self.assertEqual(
|
|
self.verifyUserData(ip_addr,configDrive_iso),
|
|
True,
|
|
"Userdata is not matching after vm reinstall"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_26_DeployVM_verify_IPFetchEvent_TC56(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm in that network
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval
|
|
5.Verify EXTERNAL.DHCP.VM.IP.FETCH event after IP address retrieval
|
|
"""
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"VM creation failed in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
#Verify IP address retrieval using listVirtualMachines API
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
#Verify the retrieved ip address in listNICs API response
|
|
self.list_nics(self.vm.id)
|
|
#Verify config drive is attached to vm or not
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm"
|
|
)
|
|
#Verify EXTERNAL.DHCP.VM.IP.FETCH event from DB
|
|
self.assertEqual(
|
|
self.verifyIPFetchEvent(ip_addr),
|
|
True,
|
|
"EXTERNAL.DHCP.VM.IP.FETCH is not generated"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_27_StopStartVM_verify_IPFetchEvent_TC57(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm in that network
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval
|
|
5.Verify EXTERNAL.DHCP.VM.IP.FETCH event after IP address retrieval
|
|
6.Stop/Start vm and repeat step5
|
|
"""
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"VM creation failed in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
#Verify IP address retrieval using listVirtualMachines API
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
#Verify the retrieved ip address in listNICs API response
|
|
self.list_nics(self.vm.id)
|
|
#Verify EXTERNAL.DHCP.VM.IP.FETCH event from DB
|
|
self.assertEqual(
|
|
self.verifyIPFetchEvent(ip_addr),
|
|
True,
|
|
"EXTERNAL.DHCP.VM.IP.FETCH is not generated"
|
|
)
|
|
#Stop/Start VM and verify IP address retrieval,event notification
|
|
self.vm.stop(self.apiclient)
|
|
self.vm.start(self.apiclient)
|
|
time.sleep(60)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm stop start"
|
|
)
|
|
#Verify EXTERNAL.DHCP.VM.IP.FETCH event from DB
|
|
self.assertEqual(
|
|
self.verifyIPFetchEvent(ip_addr),
|
|
True,
|
|
"EXTERNAL.DHCP.VM.IP.FETCH is not generated"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_28_RebootVM_verify_IPFetchEvent_TC58(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm in that network
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval
|
|
5.Verify EXTERNAL.DHCP.VM.IP.FETCH event after IP address retrieval
|
|
6.Reboot vm and repeat step5
|
|
"""
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"VM creation failed in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
#Verify IP address retrieval using listVirtualMachines API
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
#Verify the retrieved ip address in listNICs API response
|
|
self.list_nics(self.vm.id)
|
|
#Verify EXTERNAL.DHCP.VM.IP.FETCH event from DB
|
|
self.assertEqual(
|
|
self.verifyIPFetchEvent(ip_addr),
|
|
True,
|
|
"EXTERNAL.DHCP.VM.IP.FETCH is not generated"
|
|
)
|
|
#Reboot VM and verify IP address retrieval,event notification
|
|
self.vm.reboot(self.apiclient)
|
|
time.sleep(60)
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address after vm stop start"
|
|
)
|
|
#Verify EXTERNAL.DHCP.VM.IP.FETCH event from DB
|
|
self.assertEqual(
|
|
self.verifyIPFetchEvent(ip_addr),
|
|
True,
|
|
"EXTERNAL.DHCP.VM.IP.FETCH is not generated"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_29_deployVM_In_Two_Networks_TC49(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Create isolated network with VR as the service provider
|
|
2.Deploy vm in both the networks
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval
|
|
5.Verify metadata
|
|
"""
|
|
self.no_isolate = NetworkOffering.create(
|
|
self.apiclient,
|
|
self.services["isolated_network_offering"]
|
|
)
|
|
self.no_isolate.update(self.apiclient, state='Enabled')
|
|
self.isolated_network = Network.create(
|
|
self.apiclient,
|
|
self.services["network"],
|
|
networkofferingid=self.no_isolate.id,
|
|
zoneid=self.zone.id,
|
|
accountid="admin",
|
|
domainid=1
|
|
)
|
|
self.cleanup.append(self.isolated_network)
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id, self.isolated_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"VM creation failed in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
#Verify IP address retrieval using listVirtualMachines API
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
#Verify the retrieved ip address in listNICs API response
|
|
self.list_nics(self.vm.id)
|
|
#Verify config drive is attached to vm or not
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_30_deployVM_In_Two_Networks_TC50(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Create another shared network with VR as the service provider
|
|
2.Deploy vm in both the networks
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval
|
|
5.Verify metadata
|
|
"""
|
|
self.services["shared_network_offering_all_services"]["specifyVlan"]="True"
|
|
self.services["shared_network_offering_all_services"]["specifyIpRanges"]="True"
|
|
self.no_shared_vr = NetworkOffering.create(
|
|
self.apiclient,
|
|
self.services["shared_network_offering_all_services"]
|
|
)
|
|
self.no_shared_vr.update(self.apiclient, state='Enabled')
|
|
self.services["shared_network"]["vlan"] = self.services["vlan"]
|
|
self.services["shared_network"]["gateway"] = self.services["gateway"]
|
|
self.services["shared_network"]["netmask"] = self.services["netmask"]
|
|
self.services["shared_network"]["startip"] = self.services["startip"]
|
|
self.services["shared_network"]["endip"] = self.services["endip"]
|
|
self.shared_network2 = Network.create(
|
|
self.apiclient,
|
|
self.services["shared_network"],
|
|
networkofferingid=self.no_shared_vr.id,
|
|
zoneid=self.zone.id,
|
|
accountid="admin",
|
|
domainid=1
|
|
)
|
|
self.cleanup.append(self.shared_network2)
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id, self.shared_network2.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"VM creation failed in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
#Verify IP address retrieval using listVirtualMachines API
|
|
ip_status,ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
#Verify the retrieved ip address in listNICs API response
|
|
self.list_nics(self.vm.id)
|
|
#Verify config drive is attached to vm or not
|
|
configDrive_iso = self.isConfigDriveIsoAttached(ip_addr)
|
|
self.assertNotEqual(
|
|
configDrive_iso,
|
|
False,
|
|
"Config drive iso is not attached to vm"
|
|
)
|
|
metadata = self.getVMMetaData(ip_addr, configDrive_iso)
|
|
self.verifyMetaData(metadata)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_31_deployVM_with_userdataByVR_TC38(self):
|
|
"""
|
|
1.Create shared network with userdata service and VR as the provider
|
|
2.Deploy vm with userdata in the above network
|
|
3.Verify vm deployment
|
|
4.Verify IP Address retrieval
|
|
"""
|
|
self.skipTest("this combination is not supported")
|
|
self.services["virtual_machine"]["userdata"] = self.userdata
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"VM creation with userdata failed in a shared network with VR as the provider"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
#Verify IP address retrieval using listVirtualMachines API
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
#Verify the retrieved ip address in listNICs API response
|
|
self.list_nics(self.vm.id)
|
|
vr_res = Router.list(
|
|
self.apiclient,
|
|
networkid=self.shared_network.id,
|
|
listAll=True
|
|
)
|
|
self.assertEqual(validateList(vr_res)[0],PASS,"List Routers returned invalid response")
|
|
vr_ip = vr_res[0].guestipaddress
|
|
ssh = self.vm.get_ssh_client(ipaddress=ip_addr)
|
|
cmd = "curl http://%s/latest/user-data" % vr_ip
|
|
res = ssh.execute(cmd)
|
|
self.assertEqual(
|
|
str(res),
|
|
self.userdata,
|
|
"Failed to get the userdata from VR in shared network"
|
|
)
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_32_restart_network_TC6(self):
|
|
"""
|
|
1.Create shared network without any services
|
|
2.Deploy vm in that network
|
|
3.Verify vm deployment
|
|
4.Verify IP address retrieval
|
|
5.Restart network and verify access to VM
|
|
"""
|
|
self.vm = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=self.template.id,
|
|
accountid="admin",
|
|
domainid=1,
|
|
serviceofferingid=self.service_offering.id,
|
|
zoneid=self.zone.id,
|
|
networkids=[self.shared_network.id],
|
|
)
|
|
self.assertIsNotNone(
|
|
self.vm,
|
|
"VM creation failed in a shared network without any services"
|
|
)
|
|
self.cleanup.append(self.vm)
|
|
#Verify IP address retrieval using listVirtualMachines API
|
|
ip_status, ip_addr = self.getVMIPAddress(self.vm.id)
|
|
self.assertEqual(
|
|
ip_status,
|
|
1,
|
|
"Failed to retrieve vm ip address"
|
|
)
|
|
#Verify the retrieved ip address in listNICs API response
|
|
self.list_nics(self.vm.id)
|
|
res = self.shared_network.restart(self.apiclient, cleanup=True)
|
|
if res.success:
|
|
self.assertEqual(
|
|
res.success,
|
|
True,
|
|
"Failed to restart shared network without any services"
|
|
)
|
|
else:
|
|
self.fail("Failed restarting shared network without any services")
|
|
return
|
|
|
|
@attr(tags=["advanced"], required_hardware='True')
|
|
def test_33_verify_config_params_TC65(self):
|
|
"""
|
|
#@desc: Validate external dhcp config parameters against invalid values
|
|
#1.Generate random values using special characters, strings and numbers
|
|
#2.Try to update the config params with those values.
|
|
#3.Verify that invalid values are not allowed
|
|
"""
|
|
params = ["externaldhcp.vmip.retrieval.interval","externaldhcp.vmip.max.retry",\
|
|
"externaldhcp.vmipFetch.threadPool.max"]
|
|
for param in params:
|
|
random_num = ''.join(random.choice(string.digits + string.punctuation + string.ascii_letters) \
|
|
for _ in range(3))
|
|
try:
|
|
Configurations.update(
|
|
self.apiclient,
|
|
name=param,
|
|
value=random_num
|
|
)
|
|
self.fail("FAIL:{} takes invalid value".format(param))
|
|
except Exception as e:
|
|
self.debug("Success: Updating dhcp config params with invalid values is not allowed\
|
|
and throws following message: %s" % e)
|
|
return
|
|
|