mirror of
https://github.com/apache/cloudstack.git
synced 2025-10-26 08:42:29 +01:00
1017 lines
40 KiB
Python
1017 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 *
|
|
|
|
|
|
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',
|
|
},
|
|
"template": {
|
|
"displaytext": 'Template from snapshot',
|
|
"name": 'Template from snapshot',
|
|
"ostype": 'CentOS 5.3 (64-bit)',
|
|
"templatefilter": 'self',
|
|
"url": "",
|
|
"hypervisor": '',
|
|
"format": '',
|
|
"isfeatured": True,
|
|
"ispublic": True,
|
|
"isextractable": True,
|
|
"passwordenabled": True,
|
|
},
|
|
"firewall_rule": {
|
|
"cidrlist" : "0.0.0.0/0",
|
|
"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["template"]["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.
|
|
|
|
builtin_info = get_builtin_template_info(self.apiclient, self.zone.id)
|
|
self.services["template"]["url"] = builtin_info[0]
|
|
self.services["template"]["hypervisor"] = builtin_info[1]
|
|
self.services["template"]["format"] = builtin_info[2]
|
|
|
|
self.debug("Registering a new template")
|
|
|
|
# Register new template
|
|
template = Template.register(
|
|
self.apiclient,
|
|
self.services["template"],
|
|
zoneid=self.zone.id,
|
|
account=self.account.name,
|
|
domainid=self.account.domainid
|
|
)
|
|
self.debug(
|
|
"Registered a template of format: %s with ID: %s" % (
|
|
self.services["template"]["format"],
|
|
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["template"]["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, in fact it's firewall rule
|
|
nat_rule = StaticNATRule.create(
|
|
self.apiclient,
|
|
self.services["firewall_rule"],
|
|
public_ip.id
|
|
)
|
|
self.debug("Created Static NAT rule for public IP ID: %s" %
|
|
public_ip.id)
|
|
self.debug("Checking IP address")
|
|
ip_response = PublicIPAddress.list(
|
|
self.apiclient,
|
|
id = public_ip.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(ip_response, list),
|
|
True,
|
|
"Check ip response returns a valid list"
|
|
)
|
|
self.assertNotEqual(
|
|
len(ip_response),
|
|
0,
|
|
"Check static NAT Rule is created"
|
|
)
|
|
self.assertTrue(
|
|
ip_response[0].isstaticnat,
|
|
"IP is not static nat enabled"
|
|
)
|
|
self.assertEqual(
|
|
ip_response[0].virtualmachineid,
|
|
self.virtual_machine.id,
|
|
"IP is not binding with the VM"
|
|
)
|
|
|
|
self.debug("Checking Firewall rule")
|
|
firewall_response = FireWallRule.list(
|
|
self.apiclient,
|
|
ipaddressid = public_ip.id,
|
|
listall = True
|
|
)
|
|
self.assertEqual(
|
|
isinstance(firewall_response, list),
|
|
True,
|
|
"Check firewall response returns a valid list"
|
|
)
|
|
self.assertNotEqual(
|
|
len(firewall_response),
|
|
0,
|
|
"Check firewall rule is created"
|
|
)
|
|
self.assertEqual(
|
|
firewall_response[0].state,
|
|
"Active",
|
|
"Firewall rule is not active"
|
|
)
|
|
self.assertEqual(
|
|
firewall_response[0].ipaddressid,
|
|
public_ip.id,
|
|
"Firewall rule is not static nat related"
|
|
)
|
|
self.assertEqual(
|
|
firewall_response[0].startport,
|
|
str(self.services["firewall_rule"]["startport"]),
|
|
"Firewall rule is not with specific port"
|
|
)
|
|
|
|
self.debug("Removed the firewall rule")
|
|
nat_rule.delete(self.apiclient)
|
|
|
|
self.debug("Checking IP address, it should still existed")
|
|
ip_response = PublicIPAddress.list(
|
|
self.apiclient,
|
|
id = public_ip.id
|
|
)
|
|
self.assertEqual(
|
|
isinstance(ip_response, list),
|
|
True,
|
|
"Check ip response returns a valid list"
|
|
)
|
|
self.assertNotEqual(
|
|
len(ip_response),
|
|
0,
|
|
"Check static NAT Rule is created"
|
|
)
|
|
self.assertTrue(
|
|
ip_response[0].isstaticnat,
|
|
"IP is not static nat enabled"
|
|
)
|
|
self.assertEqual(
|
|
ip_response[0].virtualmachineid,
|
|
self.virtual_machine.id,
|
|
"IP is not binding with the VM"
|
|
)
|
|
|
|
self.debug("Checking Firewall rule, it should be removed")
|
|
firewall_response = FireWallRule.list(
|
|
self.apiclient,
|
|
ipaddressid = public_ip.id,
|
|
listall = True
|
|
)
|
|
self.assertEqual(
|
|
isinstance(firewall_response, list),
|
|
True,
|
|
"Check firewall response returns a valid list"
|
|
)
|
|
if len(firewall_response) != 0 :
|
|
self.assertEqual(
|
|
firewall_response[0].state,
|
|
"Deleting",
|
|
"Firewall rule should be deleted or in deleting state"
|
|
)
|
|
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):
|
|
# No need
|
|
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["template"],
|
|
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["template"]
|
|
)
|
|
self.cleanup.append(template)
|
|
|
|
self.debug("Created template from snapshot with ID: %s" % template.id)
|
|
templates = Template.list(
|
|
self.apiclient,
|
|
templatefilter=\
|
|
self.services["template"]["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,
|
|
"Derived template size (%s) does not match snapshot size (%s)" % (templates[0].size, self.volume.size)
|
|
)
|
|
return
|
|
|
|
@attr(speed = "slow")
|
|
@attr(tags = ["advanced", "advancedns", "basic", "sg", "eip"])
|
|
def test_03_reuse_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["template"],
|
|
random_name=False
|
|
)
|
|
self.debug("Created template from snapshot: %s" % template.id)
|
|
templates = Template.list(
|
|
self.apiclient,
|
|
templatefilter=\
|
|
self.services["template"]["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["template"],
|
|
random_name=False
|
|
)
|
|
|
|
templates = Template.list(
|
|
self.apiclient,
|
|
templatefilter=\
|
|
self.services["template"]["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["template"]["name"],
|
|
"Check the name of the template"
|
|
)
|
|
return
|