mirror of
https://github.com/apache/cloudstack.git
synced 2025-10-26 08:42:29 +01:00
5b238fa4af
This result in countless test case failure because many test case units didn't clean up after it. The regression test result should be improved quite a bit after this change.
1004 lines
40 KiB
Python
1004 lines
40 KiB
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 Blocker bugs
|
|
"""
|
|
import marvin
|
|
from nose.plugins.attrib import attr
|
|
from marvin.integration.lib.base import *
|
|
from marvin.integration.lib.utils import *
|
|
from marvin.integration.lib.common import *
|
|
|
|
#Import Local Modules
|
|
from marvin.cloudstackTestCase import *
|
|
from marvin.cloudstackAPI import *
|
|
from marvin.remoteSSHClient import remoteSSHClient
|
|
|
|
|
|
class Services:
|
|
"""Test Services
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.services = {
|
|
"domain": {
|
|
"name": "Domain",
|
|
},
|
|
"account": {
|
|
"email": "test@test.com",
|
|
"firstname": "Test",
|
|
"lastname": "User",
|
|
"username": "test",
|
|
# Random characters are appended for unique
|
|
# username
|
|
"password": "password",
|
|
},
|
|
"service_offering": {
|
|
"name": "Tiny Instance",
|
|
"displaytext": "Tiny Instance",
|
|
"cpunumber": 1,
|
|
"cpuspeed": 100, # in MHz
|
|
"memory": 128, # In MBs
|
|
},
|
|
"disk_offering": {
|
|
"displaytext": "Small",
|
|
"name": "Small",
|
|
"disksize": 1
|
|
},
|
|
"virtual_machine": {
|
|
"displayname": "Test VM",
|
|
"username": "root",
|
|
"password": "password",
|
|
"ssh_port": 22,
|
|
"hypervisor": 'XenServer',
|
|
"privateport": 22,
|
|
"publicport": 22,
|
|
"protocol": 'TCP',
|
|
},
|
|
"templates": {
|
|
"displaytext": 'Template from snapshot',
|
|
"name": 'Template from snapshot',
|
|
"ostype": 'CentOS 5.3 (64-bit)',
|
|
"templatefilter": 'self',
|
|
"url": "http://download.cloud.com/releases/2.0.0/UbuntuServer-10-04-64bit.vhd.bz2",
|
|
"hypervisor": 'XenServer',
|
|
"format": 'VHD',
|
|
"isfeatured": True,
|
|
"ispublic": True,
|
|
"isextractable": True,
|
|
"passwordenabled": True,
|
|
},
|
|
"static_nat": {
|
|
"startport": 22,
|
|
"endport": 22,
|
|
"protocol": "TCP"
|
|
},
|
|
"ostype": 'CentOS 5.3 (64-bit)',
|
|
# Cent OS 5.3 (64 bit)
|
|
"sleep": 60,
|
|
}
|
|
|
|
|
|
class TestTemplate(cloudstackTestCase):
|
|
|
|
def setUp(self):
|
|
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
try:
|
|
cleanup_resources(self.apiclient, self.cleanup)
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
return
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.services = Services().services
|
|
cls.api_client = super(TestTemplate, cls).getClsTestClient().getApiClient()
|
|
|
|
# Get Zone, Domain and templates
|
|
cls.domain = get_domain(cls.api_client, cls.services)
|
|
cls.zone = get_zone(cls.api_client, cls.services)
|
|
cls.services['mode'] = cls.zone.networktype
|
|
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
|
|
cls.services["templates"]["zoneid"] = cls.zone.id
|
|
|
|
cls.service_offering = ServiceOffering.create(
|
|
cls.api_client,
|
|
cls.services["service_offering"]
|
|
)
|
|
cls.account = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
domainid=cls.domain.id
|
|
)
|
|
cls.services["account"] = cls.account.name
|
|
|
|
cls._cleanup = [
|
|
cls.account,
|
|
cls.service_offering
|
|
]
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
try:
|
|
cls.api_client = super(TestTemplate, cls).getClsTestClient().getApiClient()
|
|
#Cleanup resources used
|
|
cleanup_resources(cls.api_client, cls._cleanup)
|
|
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
|
|
return
|
|
|
|
@attr(tags = ["advanced", "advancedns", "basic", "sg"])
|
|
def test_01_create_template(self):
|
|
"""TS_BUG_002-Test to create and deploy VM using password enabled template
|
|
"""
|
|
|
|
|
|
# Validate the following:
|
|
#1. Create a password enabled template
|
|
#2. Deploy VM using this template
|
|
#3. Deploy VM should return password set in template.
|
|
|
|
self.debug("Registering a new template")
|
|
# Register new template
|
|
template = Template.register(
|
|
self.apiclient,
|
|
self.services["templates"],
|
|
zoneid=self.zone.id,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.debug("Registering template with ID: %s" % template.id)
|
|
try:
|
|
# Wait for template to download
|
|
template.download(self.apiclient)
|
|
except Exception as e:
|
|
self.fail("Exception while downloading template %s: %s"\
|
|
% (template.id, e))
|
|
|
|
self.cleanup.append(template)
|
|
|
|
# Wait for template status to be changed across
|
|
time.sleep(self.services["sleep"])
|
|
|
|
list_template_response = Template.list(
|
|
self.apiclient,
|
|
templatefilter=\
|
|
self.services["templates"]["templatefilter"],
|
|
id=template.id,
|
|
zoneid=self.zone.id
|
|
)
|
|
|
|
self.assertEqual(
|
|
isinstance(list_template_response, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
#Verify template response to check whether template added successfully
|
|
self.assertNotEqual(
|
|
len(list_template_response),
|
|
0,
|
|
"Check template available in List Templates"
|
|
)
|
|
template_response = list_template_response[0]
|
|
|
|
self.assertEqual(
|
|
template_response.isready,
|
|
True,
|
|
"Check display text of newly created template"
|
|
)
|
|
|
|
# Deploy new virtual machine using template
|
|
virtual_machine = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
templateid=template.id,
|
|
accountid=self.account.name,
|
|
domainid=self.account.domainid,
|
|
serviceofferingid=self.service_offering.id,
|
|
)
|
|
self.debug("Deployed VM with ID: %s " % virtual_machine.id)
|
|
self.assertEqual(
|
|
hasattr(virtual_machine, "password"),
|
|
True,
|
|
"Check if the deployed VM returned a password"
|
|
)
|
|
return
|
|
|
|
|
|
class TestNATRules(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
|
|
cls.api_client = super(TestNATRules, cls).getClsTestClient().getApiClient()
|
|
cls.services = Services().services
|
|
|
|
# Get Zone, Domain and templates
|
|
cls.domain = get_domain(cls.api_client, cls.services)
|
|
cls.zone = get_zone(cls.api_client, cls.services)
|
|
cls.services['mode'] = cls.zone.networktype
|
|
template = get_template(
|
|
cls.api_client,
|
|
cls.zone.id,
|
|
cls.services["ostype"]
|
|
)
|
|
#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.services["virtual_machine"]["zoneid"] = cls.zone.id
|
|
cls.service_offering = ServiceOffering.create(
|
|
cls.api_client,
|
|
cls.services["service_offering"]
|
|
)
|
|
cls.virtual_machine = VirtualMachine.create(
|
|
cls.api_client,
|
|
cls.services["virtual_machine"],
|
|
templateid=template.id,
|
|
accountid=cls.account.name,
|
|
domainid=cls.account.domainid,
|
|
serviceofferingid=cls.service_offering.id
|
|
)
|
|
cls.public_ip = PublicIPAddress.create(
|
|
cls.api_client,
|
|
accountid=cls.account.name,
|
|
zoneid=cls.zone.id,
|
|
domainid=cls.account.domainid,
|
|
services=cls.services["virtual_machine"]
|
|
)
|
|
cls._cleanup = [
|
|
cls.virtual_machine,
|
|
cls.account,
|
|
cls.service_offering
|
|
]
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
try:
|
|
cleanup_resources(self.apiclient, self.cleanup)
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
try:
|
|
cls.api_client = super(TestNATRules, cls).getClsTestClient().getApiClient()
|
|
cleanup_resources(cls.api_client, cls._cleanup)
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
|
|
@attr(tags = ["advanced"])
|
|
def test_01_firewall_rules_port_fw(self):
|
|
""""Checking firewall rules deletion after static NAT disable"""
|
|
|
|
|
|
# Validate the following:
|
|
#1. Enable static NAT for a VM
|
|
#2. Open up some ports. At this point there will be new rows in the
|
|
# firewall_rules table.
|
|
#3. Disable static NAT for the VM.
|
|
#4. Check fire wall rules are deleted from firewall_rules table.
|
|
|
|
public_ip = self.public_ip.ipaddress
|
|
|
|
# Enable Static NAT for VM
|
|
StaticNATRule.enable(
|
|
self.apiclient,
|
|
public_ip.id,
|
|
self.virtual_machine.id
|
|
)
|
|
self.debug("Enabled static NAT for public IP ID: %s" %
|
|
public_ip.id)
|
|
#Create Static NAT rule
|
|
nat_rule = StaticNATRule.create(
|
|
self.apiclient,
|
|
self.services["static_nat"],
|
|
public_ip.id
|
|
)
|
|
self.debug("Created Static NAT rule for public IP ID: %s" %
|
|
public_ip.id)
|
|
list_rules_repsonse = StaticNATRule.list(
|
|
self.apiclient,
|
|
id=nat_rule.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(list_rules_repsonse, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
self.assertNotEqual(
|
|
len(list_rules_repsonse),
|
|
0,
|
|
"Check IP Forwarding Rule is created"
|
|
)
|
|
self.assertEqual(
|
|
list_rules_repsonse[0].id,
|
|
nat_rule.id,
|
|
"Check Correct IP forwarding Rule is returned"
|
|
)
|
|
# Verify the entries made in firewall_rules tables
|
|
self.debug(
|
|
"select id from user_ip_address where uuid = '%s';" \
|
|
% public_ip.id
|
|
)
|
|
qresultset = self.dbclient.execute(
|
|
"select id from user_ip_address where uuid = '%s';" \
|
|
% public_ip.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(qresultset, list),
|
|
True,
|
|
"Check database query returns a valid data"
|
|
)
|
|
|
|
self.assertNotEqual(
|
|
len(qresultset),
|
|
0,
|
|
"Check DB Query result set"
|
|
)
|
|
qresult = qresultset[0]
|
|
public_ip_id = qresult[0]
|
|
# Verify the entries made in firewall_rules tables
|
|
self.debug(
|
|
"select id, state from firewall_rules where ip_address_id = '%s';" \
|
|
% public_ip_id
|
|
)
|
|
qresultset = self.dbclient.execute(
|
|
"select id, state from firewall_rules where ip_address_id = '%s';" \
|
|
% public_ip_id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(qresultset, list),
|
|
True,
|
|
"Check database query returns a valid data for firewall rules"
|
|
)
|
|
|
|
self.assertNotEqual(
|
|
len(qresultset),
|
|
0,
|
|
"Check DB Query result set"
|
|
)
|
|
|
|
for qresult in qresultset:
|
|
self.assertEqual(
|
|
qresult[1],
|
|
'Active',
|
|
"Check state of the static NAT rule in database"
|
|
)
|
|
|
|
nat_rule.delete(self.apiclient)
|
|
|
|
list_rules_repsonse = StaticNATRule.list(
|
|
self.apiclient,
|
|
id=nat_rule.id
|
|
)
|
|
|
|
self.assertEqual(
|
|
list_rules_repsonse,
|
|
None,
|
|
"Check Port Forwarding Rule is deleted"
|
|
)
|
|
|
|
# Verify the entries made in firewall_rules tables
|
|
self.debug(
|
|
"select id, state from firewall_rules where ip_address_id = '%s';" \
|
|
% public_ip.id
|
|
)
|
|
qresultset = self.dbclient.execute(
|
|
"select id, state from firewall_rules where ip_address_id = '%s';" \
|
|
% public_ip.id
|
|
)
|
|
|
|
self.assertEqual(
|
|
len(qresultset),
|
|
0,
|
|
"Check DB Query result set"
|
|
)
|
|
return
|
|
|
|
|
|
class TestRouters(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.api_client = super(TestRouters, cls).getClsTestClient().getApiClient()
|
|
cls.services = Services().services
|
|
# Get Zone, Domain and templates
|
|
cls.zone = get_zone(cls.api_client, cls.services)
|
|
cls.services['mode'] = cls.zone.networktype
|
|
cls.template = get_template(
|
|
cls.api_client,
|
|
cls.zone.id,
|
|
cls.services["ostype"]
|
|
)
|
|
|
|
# Create an account, domain etc
|
|
cls.domain = Domain.create(
|
|
cls.api_client,
|
|
cls.services["domain"],
|
|
)
|
|
cls.admin_account = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
admin=True,
|
|
domainid=cls.domain.id
|
|
)
|
|
|
|
cls.user_account = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
domainid=cls.domain.id
|
|
)
|
|
|
|
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
|
|
cls.services["virtual_machine"]["template"] = cls.template.id
|
|
|
|
cls.service_offering = ServiceOffering.create(
|
|
cls.api_client,
|
|
cls.services["service_offering"]
|
|
)
|
|
|
|
cls._cleanup = [
|
|
cls.service_offering,
|
|
cls.admin_account,
|
|
cls.user_account,
|
|
cls.domain
|
|
]
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
try:
|
|
#Cleanup resources used
|
|
cleanup_resources(cls.api_client, cls._cleanup)
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
return
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
try:
|
|
cleanup_resources(self.apiclient, self.cleanup)
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
return
|
|
|
|
@attr(tags = ["advanced", "advancedns"])
|
|
def test_01_list_routers_admin(self):
|
|
"""TS_BUG_007-Check listRouters() using Admin User
|
|
"""
|
|
|
|
|
|
# Validate the following
|
|
# 1. PreReq: have rounters that are owned by other account
|
|
# 2. Create domain and create accounts in that domain
|
|
# 3. Create one VM for each account
|
|
# 4. Using Admin , run listRouters. It should return all the routers
|
|
|
|
vm_1 = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
accountid=self.admin_account.name,
|
|
domainid=self.admin_account.domainid,
|
|
serviceofferingid=self.service_offering.id
|
|
)
|
|
self.debug("Deployed VM with ID: %s" % vm_1.id)
|
|
vm_2 = VirtualMachine.create(
|
|
self.apiclient,
|
|
self.services["virtual_machine"],
|
|
accountid=self.user_account.name,
|
|
domainid=self.user_account.domainid,
|
|
serviceofferingid=self.service_offering.id
|
|
)
|
|
self.debug("Deployed VM with ID: %s" % vm_2.id)
|
|
routers = list_routers(
|
|
self.apiclient,
|
|
account=self.admin_account.name,
|
|
domainid=self.admin_account.domainid,
|
|
)
|
|
self.assertEqual(
|
|
isinstance(routers, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
# ListRouter Should return 2 records
|
|
self.assertEqual(
|
|
len(routers),
|
|
1,
|
|
"Check list router response"
|
|
)
|
|
for router in routers:
|
|
self.assertEqual(
|
|
router.state,
|
|
'Running',
|
|
"Check list router response for router state"
|
|
)
|
|
return
|
|
|
|
|
|
class TestRouterRestart(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
|
|
cls.api_client = super(TestRouterRestart, cls).getClsTestClient().getApiClient()
|
|
cls.services = Services().services
|
|
# Get Zone, Domain and templates
|
|
cls.domain = get_domain(cls.api_client, cls.services)
|
|
cls.zone = get_zone(cls.api_client, cls.services)
|
|
cls.services['mode'] = cls.zone.networktype
|
|
template = get_template(
|
|
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
|
|
cls.account = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
domainid=cls.domain.id
|
|
)
|
|
cls.service_offering = ServiceOffering.create(
|
|
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.cleanup = [
|
|
cls.vm_1,
|
|
cls.account,
|
|
cls.service_offering
|
|
]
|
|
return
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
try:
|
|
cls.api_client = super(TestRouterRestart, cls).getClsTestClient().getApiClient()
|
|
#Clean up, terminate the created templates
|
|
cleanup_resources(cls.api_client, cls.cleanup)
|
|
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
return
|
|
|
|
def setUp(self):
|
|
self.apiclient = self.testClient.getApiClient()
|
|
return
|
|
|
|
def tearDown(self):
|
|
try:
|
|
cleanup_resources(self.apiclient, self.cleanup)
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
return
|
|
|
|
@attr(tags = ["advanced", "basic", "sg", "advancedns", "eip"])
|
|
def test_01_restart_network_cleanup(self):
|
|
"""TS_BUG_008-Test restart network
|
|
"""
|
|
|
|
|
|
# Validate the following
|
|
# 1. When cleanup = true, router is destroyed and a new one created
|
|
# 2. New router will have new publicIp and linkLocalIp and
|
|
# all it's services should resume
|
|
|
|
# Find router associated with user account
|
|
list_router_response = list_routers(
|
|
self.apiclient,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.assertEqual(
|
|
isinstance(list_router_response, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
router = list_router_response[0]
|
|
|
|
#Store old values before restart
|
|
old_linklocalip = router.linklocalip
|
|
|
|
timeout = 10
|
|
# Network should be in Implemented or Setup stage before restart
|
|
while True:
|
|
networks = Network.list(
|
|
self.apiclient,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
network = networks[0]
|
|
if network.state in ["Implemented", "Setup"]:
|
|
break
|
|
elif timeout == 0:
|
|
break
|
|
else:
|
|
time.sleep(60)
|
|
timeout = timeout - 1
|
|
|
|
self.debug("Restarting network: %s" % network.id)
|
|
cmd = restartNetwork.restartNetworkCmd()
|
|
cmd.id = network.id
|
|
cmd.cleanup = True
|
|
self.apiclient.restartNetwork(cmd)
|
|
|
|
# Get router details after restart
|
|
list_router_response = list_routers(
|
|
self.apiclient,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.assertEqual(
|
|
isinstance(list_router_response, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
router = list_router_response[0]
|
|
|
|
self.assertNotEqual(
|
|
router.linklocalip,
|
|
old_linklocalip,
|
|
"Check link-local IP after restart"
|
|
)
|
|
return
|
|
|
|
|
|
class TestTemplates(cloudstackTestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
|
|
cls.services = Services().services
|
|
cls.api_client = super(TestTemplates, cls).getClsTestClient().getApiClient()
|
|
|
|
# Get Zone, templates etc
|
|
cls.domain = get_domain(cls.api_client, cls.services)
|
|
cls.zone = get_zone(cls.api_client, cls.services)
|
|
cls.services['mode'] = cls.zone.networktype
|
|
|
|
template = get_template(
|
|
cls.api_client,
|
|
cls.zone.id,
|
|
cls.services["ostype"]
|
|
)
|
|
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
|
|
cls.account = Account.create(
|
|
cls.api_client,
|
|
cls.services["account"],
|
|
domainid=cls.domain.id
|
|
)
|
|
|
|
cls.services["account"] = cls.account.name
|
|
cls.service_offering = ServiceOffering.create(
|
|
cls.api_client,
|
|
cls.services["service_offering"]
|
|
)
|
|
|
|
# create virtual machine
|
|
cls.virtual_machine = VirtualMachine.create(
|
|
cls.api_client,
|
|
cls.services["virtual_machine"],
|
|
templateid=template.id,
|
|
accountid=cls.account.name,
|
|
domainid=cls.account.domainid,
|
|
serviceofferingid=cls.service_offering.id,
|
|
)
|
|
#Stop virtual machine
|
|
cls.virtual_machine.stop(cls.api_client)
|
|
|
|
#Wait before server has be successfully stopped
|
|
time.sleep(30)
|
|
list_volume = Volume.list(
|
|
cls.api_client,
|
|
virtualmachineid=cls.virtual_machine.id,
|
|
type='ROOT',
|
|
listall=True
|
|
)
|
|
try:
|
|
if isinstance(list_volume, list):
|
|
cls.volume = list_volume[0]
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during setup : %s" % e)
|
|
|
|
cls._cleanup = [
|
|
cls.service_offering,
|
|
cls.account,
|
|
]
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
try:
|
|
cls.api_client = super(TestTemplates, cls).getClsTestClient().getApiClient()
|
|
#Cleanup created resources such as templates and VMs
|
|
cleanup_resources(cls.api_client, cls._cleanup)
|
|
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
|
|
return
|
|
|
|
def setUp(self):
|
|
|
|
self.apiclient = self.testClient.getApiClient()
|
|
self.dbclient = self.testClient.getDbConnection()
|
|
self.cleanup = []
|
|
return
|
|
|
|
def tearDown(self):
|
|
try:
|
|
#Clean up, terminate the created templates
|
|
cleanup_resources(self.apiclient, self.cleanup)
|
|
|
|
except Exception as e:
|
|
raise Exception("Warning: Exception during cleanup : %s" % e)
|
|
|
|
return
|
|
|
|
@attr(speed = "slow")
|
|
@attr(tags = ["advanced", "advancedns", "basic", "sg", "eip"])
|
|
def test_01_check_template_size(self):
|
|
"""TS_BUG_009-Test the size of template created from root disk
|
|
"""
|
|
|
|
|
|
# Validate the following:
|
|
# 1. Deploy new VM using the template created from Volume
|
|
# 2. VM should be in Up and Running state
|
|
|
|
#Create template from volume
|
|
template = Template.create(
|
|
self.apiclient,
|
|
self.services["templates"],
|
|
self.volume.id,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.debug("Creating template with ID: %s" % template.id)
|
|
# Volume and Template Size should be same
|
|
self.assertEqual(
|
|
template.size,
|
|
self.volume.size,
|
|
"Check if size of template and volume are same"
|
|
)
|
|
return
|
|
|
|
@attr(speed = "slow")
|
|
@attr(tags = ["advanced", "advancedns", "basic", "sg", "eip"])
|
|
def test_02_check_size_snapshotTemplate(self):
|
|
"""TS_BUG_010-Test check size of snapshot and template
|
|
"""
|
|
|
|
|
|
# Validate the following
|
|
# 1. Deploy VM using default template, small service offering
|
|
# and small data disk offering.
|
|
# 2. Perform snapshot on the root disk of this VM.
|
|
# 3. Create a template from snapshot.
|
|
# 4. Check the size of snapshot and template
|
|
|
|
# Create a snapshot from the ROOTDISK
|
|
snapshot = Snapshot.create(
|
|
self.apiclient,
|
|
self.volume.id,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.debug("Created snapshot with ID: %s" % snapshot.id)
|
|
snapshots = Snapshot.list(
|
|
self.apiclient,
|
|
id=snapshot.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(snapshots, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
self.assertNotEqual(
|
|
snapshots,
|
|
None,
|
|
"Check if result exists in list snapshots call"
|
|
)
|
|
self.assertEqual(
|
|
snapshots[0].id,
|
|
snapshot.id,
|
|
"Check snapshot id in list resources call"
|
|
)
|
|
|
|
# Generate template from the snapshot
|
|
template = Template.create_from_snapshot(
|
|
self.apiclient,
|
|
snapshot,
|
|
self.services["templates"]
|
|
)
|
|
self.cleanup.append(template)
|
|
|
|
self.debug("Created template from snapshot with ID: %s" % template.id)
|
|
templates = Template.list(
|
|
self.apiclient,
|
|
templatefilter=\
|
|
self.services["templates"]["templatefilter"],
|
|
id=template.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(templates, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
self.assertNotEqual(
|
|
templates,
|
|
None,
|
|
"Check if result exists in list item call"
|
|
)
|
|
|
|
self.assertEqual(
|
|
templates[0].isready,
|
|
True,
|
|
"Check new template state in list templates call"
|
|
)
|
|
# check size of template with that of snapshot
|
|
self.assertEqual(
|
|
templates[0].size,
|
|
self.volume.size,
|
|
"Check if size of snapshot and template matches"
|
|
)
|
|
return
|
|
|
|
@attr(speed = "slow")
|
|
@attr(tags = ["advanced", "advancedns", "basic", "sg", "eip"])
|
|
def test_03_resuse_template_name(self):
|
|
"""TS_BUG_011-Test Reusing deleted template name
|
|
"""
|
|
|
|
|
|
# Validate the following
|
|
# 1. Deploy VM using default template, small service offering
|
|
# and small data disk offering.
|
|
# 2. Perform snapshot on the root disk of this VM.
|
|
# 3. Create a template from snapshot.
|
|
# 4. Delete the template and create a new template with same name
|
|
# 5. Template should be created succesfully
|
|
|
|
# Create a snapshot from the ROOTDISK
|
|
snapshot = Snapshot.create(
|
|
self.apiclient,
|
|
self.volume.id,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.debug("Created snapshot with ID: %s" % snapshot.id)
|
|
snapshots = Snapshot.list(
|
|
self.apiclient,
|
|
id=snapshot.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(snapshots, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
self.assertNotEqual(
|
|
snapshots,
|
|
None,
|
|
"Check if result exists in list snapshots call"
|
|
)
|
|
self.assertEqual(
|
|
snapshots[0].id,
|
|
snapshot.id,
|
|
"Check snapshot id in list resources call"
|
|
)
|
|
|
|
# Generate template from the snapshot
|
|
template = Template.create_from_snapshot(
|
|
self.apiclient,
|
|
snapshot,
|
|
self.services["templates"],
|
|
random_name=False
|
|
)
|
|
self.debug("Created template from snapshot: %s" % template.id)
|
|
templates = Template.list(
|
|
self.apiclient,
|
|
templatefilter=\
|
|
self.services["templates"]["templatefilter"],
|
|
id=template.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(templates, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
self.assertNotEqual(
|
|
templates,
|
|
None,
|
|
"Check if result exists in list item call"
|
|
)
|
|
|
|
self.assertEqual(
|
|
templates[0].isready,
|
|
True,
|
|
"Check new template state in list templates call"
|
|
)
|
|
|
|
self.debug("Deleting template: %s" % template.id)
|
|
template.delete(self.apiclient)
|
|
|
|
# Wait for some time to ensure template state is reflected in other calls
|
|
time.sleep(self.services["sleep"])
|
|
|
|
# Generate template from the snapshot
|
|
self.debug("Creating template from snapshot: %s with same name" %
|
|
template.id)
|
|
template = Template.create_from_snapshot(
|
|
self.apiclient,
|
|
snapshot,
|
|
self.services["templates"],
|
|
random_name=False
|
|
)
|
|
|
|
templates = Template.list(
|
|
self.apiclient,
|
|
templatefilter=\
|
|
self.services["templates"]["templatefilter"],
|
|
id=template.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(templates, list),
|
|
True,
|
|
"Check list response returns a valid list"
|
|
)
|
|
self.assertNotEqual(
|
|
templates,
|
|
None,
|
|
"Check if result exists in list item call"
|
|
)
|
|
|
|
self.assertEqual(
|
|
templates[0].name,
|
|
self.services["templates"]["name"],
|
|
"Check the name of the template"
|
|
)
|
|
return
|