OpenStack Orchestration (Heat)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

5350 lines
240 KiB

#
# Licensed 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.
import collections
import contextlib
import copy
from unittest import mock
from keystoneauth1 import exceptions as ks_exceptions
from neutronclient.v2_0 import client as neutronclient
from novaclient import exceptions as nova_exceptions
from oslo_serialization import jsonutils
from oslo_utils import uuidutils
import requests
from urllib import parse as urlparse
from heat.common import exception
from heat.common.i18n import _
from heat.common import template_format
from heat.engine.clients.os import glance
from heat.engine.clients.os import heat_plugin
from heat.engine.clients.os import neutron
from heat.engine.clients.os import nova
from heat.engine.clients.os import swift
from heat.engine.clients.os import zaqar
from heat.engine import environment
from heat.engine import resource
from heat.engine.resources.openstack.nova import server as servers
from heat.engine.resources.openstack.nova import server_network_mixin
from heat.engine.resources import scheduler_hints as sh
from heat.engine import scheduler
from heat.engine import stack as parser
from heat.engine import template
from heat.objects import resource_data as resource_data_object
from heat.tests import common
from heat.tests.openstack.nova import fakes as fakes_nova
from heat.tests import utils
wp_template = '''
{
"AWSTemplateFormatVersion" : "2010-09-09",
"Description" : "WordPress",
"Parameters" : {
"key_name" : {
"Description" : "key_name",
"Type" : "String",
"Default" : "test"
}
},
"Resources" : {
"WebServer": {
"Type": "OS::Nova::Server",
"Properties": {
"image" : "F18-x86_64-gold",
"flavor" : "m1.large",
"key_name" : "test",
"user_data" : "wordpress"
}
}
}
}
'''
ns_template = '''
heat_template_version: 2015-04-30
resources:
server:
type: OS::Nova::Server
properties:
image: F17-x86_64-gold
flavor: m1.large
user_data: {get_file: a_file}
networks: [{'network': 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'}]
'''
with_port_template = '''
heat_template_version: 2015-04-30
resources:
port:
type: OS::Neutron::Port
properties:
network: 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
server:
type: OS::Nova::Server
properties:
image: F17-x86_64-gold
flavor: m1.small
networks:
- port: {get_resource: port}
fixed_ip: 10.0.0.99
'''
bdm_v2_template = '''
heat_template_version: 2015-04-30
resources:
server:
type: OS::Nova::Server
properties:
flavor: m1.tiny
block_device_mapping_v2:
- device_name: vda
delete_on_termination: true
image_id: F17-x86_64-gold
'''
subnet_template = '''
heat_template_version: 2013-05-23
resources:
server:
type: OS::Nova::Server
properties:
image: F17-x86_64-gold
flavor: m1.large
networks:
- { network: 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa' }
subnet:
type: OS::Neutron::Subnet
properties:
network: 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
subnet_unreferenced:
type: OS::Neutron::Subnet
properties:
network: 'bbccbbcc-bbcc-bbcc-bbcc-bbccbbccbbcc'
'''
multi_subnet_template = '''
heat_template_version: 2013-05-23
resources:
server:
type: OS::Nova::Server
properties:
image: F17-x86_64-gold
flavor: m1.large
networks:
- network: {get_resource: network}
network:
type: OS::Neutron::Net
properties:
name: NewNetwork
subnet1:
type: OS::Neutron::Subnet
properties:
network: {get_resource: network}
name: NewSubnet1
subnet2:
type: OS::Neutron::Subnet
properties:
network: {get_resource: network}
name: NewSubnet2
'''
no_subnet_template = '''
heat_template_version: 2013-05-23
resources:
server:
type: OS::Nova::Server
properties:
image: F17-x86_64-gold
flavor: m1.large
subnet:
type: OS::Neutron::Subnet
properties:
network: 12345
'''
tmpl_server_with_network_id = """
heat_template_version: 2015-10-15
resources:
server:
type: OS::Nova::Server
properties:
flavor: m1.small
image: F17-x86_64-gold
networks:
- network: 4321
"""
tmpl_server_with_sub_secu_group = """
heat_template_version: 2015-10-15
resources:
server:
type: OS::Nova::Server
properties:
flavor: m1.small
image: F17-x86_64-gold
networks:
- subnet: 2a60cbaa-3d33-4af6-a9ce-83594ac546fc
security_groups:
- my_seg
"""
server_with_sw_config_personality = """
heat_template_version: 2014-10-16
resources:
swconfig:
type: OS::Heat::SoftwareConfig
properties:
config: |
#!/bin/bash
echo -e "test"
server:
type: OS::Nova::Server
properties:
image: F17-x86_64-gold
flavor: m1.large
personality: { /tmp/test: { get_attr: [swconfig, config]}}
"""
def create_fake_iface(port=None, net=None, mac=None, ip=None, subnet=None):
class fake_interface(object):
def __init__(self, port_id, net_id, mac_addr, fixed_ip, subnet_id):
self.port_id = port_id
self.net_id = net_id
self.mac_addr = mac_addr
self.fixed_ips = [{'ip_address': fixed_ip, 'subnet_id': subnet_id}]
return fake_interface(port, net, mac, ip, subnet)
class ServerStatus(object):
def __init__(self, server, statuses):
self._server = server
self._status = iter(statuses)
def __call__(self, server_id):
try:
self._server.status = next(self._status)
except StopIteration:
raise AssertionError('Unexpected call to servers.get()')
return self._server
class ServersTest(common.HeatTestCase):
def setUp(self):
super(ServersTest, self).setUp()
self.fc = fakes_nova.FakeClient()
self.limits = mock.Mock()
self.limits.absolute = self._limits_absolute()
self.mock_flavor = mock.Mock(ram=4, disk=4)
self.mock_image = mock.Mock(min_ram=1, min_disk=1, status='ACTIVE')
def flavor_side_effect(*args):
return 2 if args[0] == 'm1.small' else 1
def image_side_effect(*args):
return 2 if args[0] == 'F17-x86_64-gold' else 1
self.patchobject(nova.NovaClientPlugin, 'find_flavor_by_name_or_id',
side_effect=flavor_side_effect)
self.patchobject(glance.GlanceClientPlugin, 'find_image_by_name_or_id',
side_effect=image_side_effect)
self.port_show = self.patchobject(neutronclient.Client,
'show_port')
self.subnet_show = self.patchobject(neutronclient.Client,
'show_subnet')
self.network_show = self.patchobject(neutronclient.Client,
'show_network')
def _limits_absolute(self):
max_personality = mock.Mock()
max_personality.name = 'maxPersonality'
max_personality.value = 5
max_personality_size = mock.Mock()
max_personality_size.name = 'maxPersonalitySize'
max_personality_size.value = 10240
max_server_meta = mock.Mock()
max_server_meta.name = 'maxServerMeta'
max_server_meta.value = 3
yield max_personality
yield max_personality_size
yield max_server_meta
def _setup_test_stack(self, stack_name, test_templ=wp_template):
t = template_format.parse(test_templ)
files = {}
if test_templ == ns_template:
files = {'a_file': 'stub'}
templ = template.Template(t,
env=environment.Environment(
{'key_name': 'test'}),
files=files)
stack = parser.Stack(utils.dummy_context(region_name="RegionOne"),
stack_name, templ,
stack_id=uuidutils.generate_uuid(),
stack_user_project_id='8888')
return templ, stack
def _prepare_server_check(self, status='ACTIVE'):
templ, self.stack = self._setup_test_stack('server_check')
server = self.fc.servers.list()[1]
server.status = status
res = self.stack['WebServer']
res.state_set(res.CREATE, res.COMPLETE)
res.client = mock.Mock()
res.client().servers.get.return_value = server
return res
def test_check(self):
res = self._prepare_server_check()
scheduler.TaskRunner(res.check)()
self.assertEqual((res.CHECK, res.COMPLETE), res.state)
def test_check_fail(self):
res = self._prepare_server_check()
res.client().servers.get.side_effect = Exception('boom')
exc = self.assertRaises(exception.ResourceFailure,
scheduler.TaskRunner(res.check))
self.assertIn('boom', str(exc))
self.assertEqual((res.CHECK, res.FAILED), res.state)
def test_check_not_active(self):
res = self._prepare_server_check(status='FOO')
exc = self.assertRaises(exception.ResourceFailure,
scheduler.TaskRunner(res.check))
self.assertIn('FOO', str(exc))
def _get_test_template(self, stack_name, server_name=None,
image_id=None):
tmpl, stack = self._setup_test_stack(stack_name)
tmpl.t['Resources']['WebServer']['Properties'][
'image'] = image_id or 'CentOS 5.2'
tmpl.t['Resources']['WebServer']['Properties'][
'flavor'] = '256 MB Server'
if server_name is not None:
tmpl.t['Resources']['WebServer']['Properties'][
'name'] = server_name
return tmpl, stack
def _setup_test_server(self, return_server, name, image_id=None,
override_name=False, stub_create=True,
networks=None):
stack_name = '%s_s' % name
def _mock_find_id(resource, name_or_id, cmd_resource=None):
return name_or_id
mock_find = self.patchobject(neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id')
mock_find.side_effect = _mock_find_id
server_name = str(name) if override_name else None
tmpl, self.stack = self._get_test_template(stack_name, server_name,
image_id)
props = tmpl.t['Resources']['WebServer']['Properties']
# set old_networks for server
if networks is not None:
props['networks'] = networks
self.server_props = props
resource_defns = tmpl.resource_definitions(self.stack)
server = servers.Server(str(name), resource_defns['WebServer'],
self.stack)
self.patchobject(server, 'store_external_ports')
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
if stub_create:
self.patchobject(self.fc.servers, 'create',
return_value=return_server)
# mock check_create_complete innards
self.patchobject(self.fc.servers, 'get',
return_value=return_server)
return server
def _create_test_server(self, return_server, name, override_name=False,
stub_create=True, networks=None):
server = self._setup_test_server(return_server, name,
stub_create=stub_create,
networks=networks)
scheduler.TaskRunner(server.create)()
return server
def test_subnet_dependency_by_network_id(self):
templ, stack = self._setup_test_stack('subnet-test',
subnet_template)
server_rsrc = stack['server']
subnet_rsrc = stack['subnet']
deps = []
server_rsrc.add_explicit_dependencies(deps)
server_rsrc.add_dependencies(deps)
self.assertEqual(4, len(deps))
self.assertEqual(subnet_rsrc, deps[3])
self.assertNotIn(stack['subnet_unreferenced'], deps)
def test_subnet_dependency_unknown_network_id(self):
# The use case here is creating a network + subnets + server
# from within one stack
templ, stack = self._setup_test_stack('subnet-test',
multi_subnet_template)
server_rsrc = stack['server']
subnet1_rsrc = stack['subnet1']
subnet2_rsrc = stack['subnet2']
deps = []
server_rsrc.add_explicit_dependencies(deps)
server_rsrc.add_dependencies(deps)
self.assertEqual(8, len(deps))
self.assertIn(subnet1_rsrc, deps)
self.assertIn(subnet2_rsrc, deps)
def test_subnet_nodeps(self):
templ, stack = self._setup_test_stack('subnet-test',
no_subnet_template)
server_rsrc = stack['server']
subnet_rsrc = stack['subnet']
deps = []
server_rsrc.add_explicit_dependencies(deps)
server_rsrc.add_dependencies(deps)
self.assertEqual(2, len(deps))
self.assertNotIn(subnet_rsrc, deps)
def test_server_create(self):
return_server = self.fc.servers.list()[1]
return_server.id = '5678'
return_server._info['os_collect_config'] = {}
server_name = 'test_server_create'
stack_name = '%s_s' % server_name
server = self._create_test_server(return_server, server_name)
self.patchobject(nova.NovaClientPlugin, 'is_version_supported',
return_value=True)
# this makes sure the auto increment worked on server creation
self.assertGreater(server.id, 0)
interfaces = [create_fake_iface(port='1234',
mac='fa:16:3e:8c:22:aa',
ip='4.5.6.7'),
create_fake_iface(port='5678',
mac='fa:16:3e:8c:33:bb',
ip='5.6.9.8'),
create_fake_iface(port='1013',
mac='fa:16:3e:8c:44:cc',
ip='10.13.12.13',
subnet='private_subnet_id')]
ports = [dict(id=interfaces[0].port_id,
mac_address=interfaces[0].mac_addr,
fixed_ips=interfaces[0].fixed_ips,
network_id='public_id'),
dict(id=interfaces[1].port_id,
mac_address=interfaces[1].mac_addr,
fixed_ips=interfaces[1].fixed_ips,
network_id='public_id'),
dict(id=interfaces[2].port_id,
mac_address=interfaces[2].mac_addr,
fixed_ips=interfaces[2].fixed_ips,
network_id='private_id')]
public_net = dict(id='public_id',
name='public',
mtu=1500,
subnets=['public_subnet_id'])
private_net = dict(id='private_id',
name='private',
mtu=1500,
subnets=['private_subnet_id'])
private_subnet = dict(id='private_subnet_id',
name='private_subnet',
cidr='private_cidr',
allocation_pools=[{'start': 'start_addr',
'end': 'end_addr'}],
gateway_ip='private_gateway',
network_id='private_id')
self.patchobject(self.fc.servers, 'get', return_value=return_server)
self.patchobject(neutronclient.Client, 'list_ports',
return_value={'ports': ports})
self.patchobject(neutronclient.Client, 'list_networks',
side_effect=[{'networks': [public_net]},
{'networks': [public_net]},
{'networks': [private_net]}])
self.patchobject(neutronclient.Client, 'list_floatingips',
return_value={'floatingips': []})
self.patchobject(self.fc.servers, 'tag_list', return_value=['test'])
self.subnet_show.return_value = {'subnet': private_subnet}
self.network_show.return_value = {'network': private_net}
public_ip = return_server.networks['public'][0]
self.assertEqual('1234',
server.FnGetAtt('addresses')['public'][0]['port'])
self.assertEqual('5678',
server.FnGetAtt('addresses')['public'][1]['port'])
self.assertEqual(public_ip,
server.FnGetAtt('addresses')['public'][0]['addr'])
self.assertEqual(public_ip,
server.FnGetAtt('networks')['public'][0])
private_ip = return_server.networks['private'][0]
self.assertEqual('1013',
server.FnGetAtt('addresses')['private'][0]['port'])
self.assertEqual(private_ip,
server.FnGetAtt('addresses')['private'][0]['addr'])
self.assertEqual([private_subnet],
server.FnGetAtt('addresses')['private'][0]['subnets'])
self.assertEqual(private_net,
server.FnGetAtt('addresses')['private'][0]['network'])
self.assertEqual(private_ip,
server.FnGetAtt('networks')['private'][0])
self.assertEqual(return_server._info, server.FnGetAtt('show'))
self.assertEqual('sample-server2', server.FnGetAtt('instance_name'))
self.assertEqual('192.0.2.0', server.FnGetAtt('accessIPv4'))
self.assertEqual('::babe:4317:0A83', server.FnGetAtt('accessIPv6'))
expected_name = utils.PhysName(stack_name, server.name)
self.assertEqual(expected_name, server.FnGetAtt('name'))
self.assertEqual(['test'], server.FnGetAtt('tags'))
# test with unsupported version
self.patchobject(nova.NovaClientPlugin, 'is_version_supported',
return_value=False)
if server.attributes._resolved_values.get('tags'):
del server.attributes._resolved_values['tags']
self.assertIsNone(server.FnGetAtt('tags'))
self.assertEqual({}, server.FnGetAtt('os_collect_config'))
def test_server_create_metadata(self):
stack_name = 'create_metadata_test_stack'
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
return_server = self.fc.servers.list()[1]
(tmpl, stack) = self._setup_test_stack(stack_name)
tmpl['Resources']['WebServer']['Properties'][
'metadata'] = {'a': 1}
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('create_metadata_test_server',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
mock_create = self.patchobject(self.fc.servers, 'create',
return_value=return_server)
scheduler.TaskRunner(server.create)()
args, kwargs = mock_create.call_args
self.assertEqual({'a': "1"}, kwargs['meta'])
def test_server_create_with_subnet_security_group(self):
stack_name = 'server_with_subnet_security_group'
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
return_server = self.fc.servers.list()[1]
(tmpl, stack) = self._setup_test_stack(
stack_name, test_templ=tmpl_server_with_sub_secu_group)
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('server_with_sub_secu',
resource_defns['server'], stack)
mock_find = self.patchobject(
neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id',
return_value='2a60cbaa-3d33-4af6-a9ce-83594ac546fc')
sec_uuids = ['86c0f8ae-23a8-464f-8603-c54113ef5467']
self.patchobject(neutron.NeutronClientPlugin,
'get_secgroup_uuids', return_value=sec_uuids)
self.patchobject(server, 'store_external_ports')
self.patchobject(neutron.NeutronClientPlugin,
'network_id_from_subnet_id',
return_value='05d8e681-4b37-4570-bc8d-810089f706b2')
mock_create_port = self.patchobject(
neutronclient.Client, 'create_port')
mock_create = self.patchobject(
self.fc.servers, 'create', return_value=return_server)
scheduler.TaskRunner(server.create)()
kwargs = {'network_id': '05d8e681-4b37-4570-bc8d-810089f706b2',
'fixed_ips': [
{'subnet_id': '2a60cbaa-3d33-4af6-a9ce-83594ac546fc'}],
'security_groups': sec_uuids,
'name': 'server_with_sub_secu-port-0',
}
mock_create_port.assert_called_with({'port': kwargs})
self.assertEqual(1, mock_find.call_count)
args, kwargs = mock_create.call_args
self.assertEqual({}, kwargs['meta'])
def test_server_create_with_str_network(self):
stack_name = 'server_with_str_network'
return_server = self.fc.servers.list()[1]
(tmpl, stack) = self._setup_test_stack(stack_name)
mock_nc = self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
self.patchobject(neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id')
props = tmpl['Resources']['WebServer']['Properties']
props['networks'] = [{'allocate_network': 'none'}]
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
create_mock = self.patchobject(self.fc.servers, 'create',
return_value=return_server)
scheduler.TaskRunner(server.create)()
mock_nc.assert_called_with()
self.assertEqual(3, mock_nc.call_count)
self.assertEqual('none', create_mock.call_args[1]['nics'])
def test_server_create_with_image_id(self):
return_server = self.fc.servers.list()[1]
return_server.id = '5678'
server_name = 'test_server_create_image_id'
server = self._setup_test_server(return_server,
server_name,
override_name=True)
server.resource_id = '1234'
interfaces = [create_fake_iface(port='1234',
mac='fa:16:3e:8c:22:aa',
ip='4.5.6.7'),
create_fake_iface(port='5678',
mac='fa:16:3e:8c:33:bb',
ip='5.6.9.8'),
create_fake_iface(port='1013',
mac='fa:16:3e:8c:44:cc',
ip='10.13.12.13')]
ports = [dict(id=interfaces[0].port_id,
mac_address=interfaces[0].mac_addr,
fixed_ips=interfaces[0].fixed_ips,
network_id='public_id'),
dict(id=interfaces[1].port_id,
mac_address=interfaces[1].mac_addr,
fixed_ips=interfaces[1].fixed_ips,
network_id='public_id'),
dict(id=interfaces[2].port_id,
mac_address=interfaces[2].mac_addr,
fixed_ips=interfaces[2].fixed_ips,
network_id='private_id')]
public_net = dict(id='public_id', name='public')
private_net = dict(id='private_id', name='private')
self.patchobject(self.fc.servers, 'get', return_value=return_server)
self.patchobject(neutronclient.Client, 'list_ports',
return_value={'ports': ports})
self.patchobject(neutronclient.Client, 'list_networks',
side_effect=[{'networks': [public_net]},
{'networks': [public_net]},
{'networks': [private_net]}])
self.patchobject(neutronclient.Client, 'list_floatingips',
return_value={'floatingips': []})
self.patchobject(return_server, 'interface_detach')
self.patchobject(return_server, 'interface_attach')
public_ip = return_server.networks['public'][0]
self.assertEqual('1234',
server.FnGetAtt('addresses')['public'][0]['port'])
self.assertEqual('5678',
server.FnGetAtt('addresses')['public'][1]['port'])
self.assertEqual(public_ip,
server.FnGetAtt('addresses')['public'][0]['addr'])
self.assertEqual(public_ip,
server.FnGetAtt('networks')['public'][0])
private_ip = return_server.networks['private'][0]
self.assertEqual('1013',
server.FnGetAtt('addresses')['private'][0]['port'])
self.assertEqual(private_ip,
server.FnGetAtt('addresses')['private'][0]['addr'])
self.assertEqual(private_ip,
server.FnGetAtt('networks')['private'][0])
self.assertEqual(server_name, server.FnGetAtt('name'))
def test_server_image_name_err(self):
stack_name = 'img_name_err'
(tmpl, stack) = self._setup_test_stack(stack_name)
mock_image = self.patchobject(glance.GlanceClientPlugin,
'find_image_by_name_or_id')
self.stub_KeypairConstraint_validate()
mock_image.side_effect = (
glance.client_exception.EntityMatchNotFound(
entity='image', args={'name': 'Slackware'}))
# Init a server with non exist image name
tmpl['Resources']['WebServer']['Properties']['image'] = 'Slackware'
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
error = self.assertRaises(exception.ResourceFailure,
scheduler.TaskRunner(server.create))
self.assertIn("No image matching {'name': 'Slackware'}.",
str(error))
def test_server_duplicate_image_name_err(self):
stack_name = 'img_dup_err'
(tmpl, stack) = self._setup_test_stack(stack_name)
mock_image = self.patchobject(glance.GlanceClientPlugin,
'find_image_by_name_or_id')
self.stub_KeypairConstraint_validate()
mock_image.side_effect = (
glance.client_exception.EntityUniqueMatchNotFound(
entity='image', args='CentOS 5.2'))
tmpl['Resources']['WebServer']['Properties']['image'] = 'CentOS 5.2'
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
error = self.assertRaises(exception.ResourceFailure,
scheduler.TaskRunner(server.create))
self.assertIn('No image unique match found for CentOS 5.2.',
str(error))
def test_server_create_unexpected_status(self):
# NOTE(pshchelo) checking is done only on check_create_complete
# level so not to mock out all delete/retry logic that kicks in
# on resource create failure
return_server = self.fc.servers.list()[1]
server = self._create_test_server(return_server,
'cr_unexp_sts')
return_server.status = 'BOGUS'
self.patchobject(self.fc.servers, 'get',
return_value=return_server)
e = self.assertRaises(exception.ResourceUnknownStatus,
server.check_create_complete,
server.resource_id)
self.assertEqual('Server is not active - Unknown status BOGUS due to '
'"Unknown"', str(e))
def test_server_create_error_status(self):
# NOTE(pshchelo) checking is done only on check_create_complete
# level so not to mock out all delete/retry logic that kicks in
# on resource create failure
return_server = self.fc.servers.list()[1]
server = self._create_test_server(return_server,
'cr_err_sts')
return_server.status = 'ERROR'
return_server.fault = {
'message': 'NoValidHost',
'code': 500,
'created': '2013-08-14T03:12:10Z'
}
self.patchobject(self.fc.servers, 'get',
return_value=return_server)
e = self.assertRaises(exception.ResourceInError,
server.check_create_complete,
server.resource_id)
self.assertEqual(
'Went to status ERROR due to "Message: NoValidHost, Code: 500"',
str(e))
def test_server_create_raw_userdata(self):
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
return_server = self.fc.servers.list()[1]
stack_name = 'raw_userdata_s'
(tmpl, stack) = self._setup_test_stack(stack_name)
tmpl['Resources']['WebServer']['Properties'][
'user_data_format'] = 'RAW'
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
mock_create = self.patchobject(self.fc.servers, 'create',
return_value=return_server)
scheduler.TaskRunner(server.create)()
args, kwargs = mock_create.call_args
self.assertEqual('wordpress', kwargs['userdata'])
self.assertEqual({}, kwargs['meta'])
def test_server_create_raw_config_userdata(self):
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
return_server = self.fc.servers.list()[1]
stack_name = 'raw_userdata_s'
(tmpl, stack) = self._setup_test_stack(stack_name)
tmpl['Resources']['WebServer']['Properties'][
'user_data_format'] = 'RAW'
tmpl['Resources']['WebServer']['Properties'][
'user_data'] = '8c813873-f6ee-4809-8eec-959ef39acb55'
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
self.rpc_client = mock.MagicMock()
server._rpc_client = self.rpc_client
sc = {'config': 'wordpress from config'}
self.rpc_client.show_software_config.return_value = sc
mock_create = self.patchobject(self.fc.servers, 'create',
return_value=return_server)
scheduler.TaskRunner(server.create)()
args, kwargs = mock_create.call_args
self.assertEqual('wordpress from config', kwargs['userdata'])
self.assertEqual({}, kwargs['meta'])
def test_server_create_raw_config_userdata_None(self):
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
return_server = self.fc.servers.list()[1]
stack_name = 'raw_userdata_s'
(tmpl, stack) = self._setup_test_stack(stack_name)
sc_id = '8c813873-f6ee-4809-8eec-959ef39acb55'
tmpl['Resources']['WebServer']['Properties'][
'user_data_format'] = 'RAW'
tmpl['Resources']['WebServer']['Properties']['user_data'] = sc_id
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
self.rpc_client = mock.MagicMock()
server._rpc_client = self.rpc_client
@contextlib.contextmanager
def exc_filter(*args):
try:
yield
except exception.NotFound:
pass
self.rpc_client.ignore_error_by_name.side_effect = exc_filter
self.rpc_client.show_software_config.side_effect = exception.NotFound
mock_create = self.patchobject(self.fc.servers, 'create',
return_value=return_server)
scheduler.TaskRunner(server.create)()
args, kwargs = mock_create.call_args
self.assertEqual(sc_id, kwargs['userdata'])
self.assertEqual({}, kwargs['meta'])
def _server_create_software_config(self, md=None,
stack_name='software_config_s',
ret_tmpl=False):
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
return_server = self.fc.servers.list()[1]
(tmpl, stack) = self._setup_test_stack(stack_name)
self.stack = stack
self.server_props = tmpl['Resources']['WebServer']['Properties']
self.server_props['user_data_format'] = 'SOFTWARE_CONFIG'
if md is not None:
tmpl['Resources']['WebServer']['Metadata'] = md
stack.stack_user_project_id = '8888'
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
self.patchobject(server, 'heat')
self.patchobject(self.fc.servers, 'create',
return_value=return_server)
scheduler.TaskRunner(server.create)()
self.assertEqual('4567', server.access_key)
self.assertEqual('8901', server.secret_key)
self.assertEqual('1234', server._get_user_id())
self.assertEqual('POLL_SERVER_CFN',
server.properties.get('software_config_transport'))
self.assertTrue(stack.access_allowed('4567', 'WebServer'))
self.assertFalse(stack.access_allowed('45678', 'WebServer'))
self.assertFalse(stack.access_allowed('4567', 'wWebServer'))
if ret_tmpl:
return server, tmpl
else:
return server
@mock.patch.object(heat_plugin.HeatClientPlugin, 'url_for')
def test_server_create_software_config(self, fake_url):
fake_url.return_value = 'http://ip:8000/v1'
server = self._server_create_software_config()
self.assertEqual({
'os-collect-config': {
'cfn': {
'access_key_id': '4567',
'metadata_url': 'http://ip:8000/v1/',
'path': 'WebServer.Metadata',
'secret_access_key': '8901',
'stack_name': 'software_config_s'
},
'collectors': ['ec2', 'cfn', 'local']
},
'deployments': []
}, server.metadata_get())
@mock.patch.object(heat_plugin.HeatClientPlugin, 'url_for')
def test_resolve_attribute_os_collect_config(self, fake_url):
fake_url.return_value = 'http://ip/heat-api-cfn/v1'
server = self._server_create_software_config()
self.assertEqual({
'cfn': {
'access_key_id': '4567',
'metadata_url': 'http://ip/heat-api-cfn/v1/',
'path': 'WebServer.Metadata',
'secret_access_key': '8901',
'stack_name': 'software_config_s'
},
'collectors': ['ec2', 'cfn', 'local']
}, server.FnGetAtt('os_collect_config'))
@mock.patch.object(heat_plugin.HeatClientPlugin, 'url_for')
def test_server_create_software_config_metadata(self, fake_url):
md = {'os-collect-config': {'polling_interval': 10}}
fake_url.return_value = 'http://ip/heat-api-cfn/v1'
server = self._server_create_software_config(md=md)
self.assertEqual({
'os-collect-config': {
'cfn': {
'access_key_id': '4567',
'metadata_url': 'http://ip/heat-api-cfn/v1/',
'path': 'WebServer.Metadata',
'secret_access_key': '8901',
'stack_name': 'software_config_s'
},
'collectors': ['ec2', 'cfn', 'local'],
'polling_interval': 10
},
'deployments': []
}, server.metadata_get())
def _server_create_software_config_poll_heat(self, md=None):
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
return_server = self.fc.servers.list()[1]
stack_name = 'software_config_s'
(tmpl, stack) = self._setup_test_stack(stack_name)
props = tmpl.t['Resources']['WebServer']['Properties']
props['user_data_format'] = 'SOFTWARE_CONFIG'
props['software_config_transport'] = 'POLL_SERVER_HEAT'
if md is not None:
tmpl.t['Resources']['WebServer']['Metadata'] = md
self.server_props = props
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
self.patchobject(self.fc.servers, 'create',
return_value=return_server)
scheduler.TaskRunner(server.create)()
self.assertEqual('1234', server._get_user_id())
self.assertTrue(stack.access_allowed('1234', 'WebServer'))
self.assertFalse(stack.access_allowed('45678', 'WebServer'))
self.assertFalse(stack.access_allowed('4567', 'wWebServer'))
return stack, server
def test_server_create_software_config_poll_heat(self):
stack, server = self._server_create_software_config_poll_heat()
self.assertEqual({
'os-collect-config': {
'heat': {
'auth_url': 'http://server.test:5000/v2.0',
'password': server.password,
'project_id': '8888',
'region_name': 'RegionOne',
'resource_name': 'WebServer',
'stack_id': 'software_config_s/%s' % stack.id,
'user_id': '1234'
},
'collectors': ['ec2', 'heat', 'local']
},
'deployments': []
}, server.metadata_get())
def test_server_create_software_config_poll_heat_metadata(self):
md = {'os-collect-config': {'polling_interval': 10}}
stack, server = self._server_create_software_config_poll_heat(md=md)
self.assertEqual({
'os-collect-config': {
'heat': {
'auth_url': 'http://server.test:5000/v2.0',
'password': server.password,
'project_id': '8888',
'region_name': 'RegionOne',
'resource_name': 'WebServer',
'stack_id': 'software_config_s/%s' % stack.id,
'user_id': '1234'
},
'collectors': ['ec2', 'heat', 'local'],
'polling_interval': 10
},
'deployments': []
}, server.metadata_get())
def _server_create_software_config_poll_temp_url(self, md=None):
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
return_server = self.fc.servers.list()[1]
stack_name = 'software_config_s'
(tmpl, stack) = self._setup_test_stack(stack_name)
props = tmpl.t['Resources']['WebServer']['Properties']
props['user_data_format'] = 'SOFTWARE_CONFIG'
props['software_config_transport'] = 'POLL_TEMP_URL'
if md is not None:
tmpl.t['Resources']['WebServer']['Metadata'] = md
self.server_props = props
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
sc = mock.Mock()
sc.head_account.return_value = {
'x-account-meta-temp-url-key': 'secrit'
}
sc.url = 'http://192.0.2.2'
self.patchobject(swift.SwiftClientPlugin, '_create',
return_value=sc)
self.patchobject(self.fc.servers, 'create',
return_value=return_server)
scheduler.TaskRunner(server.create)()
metadata_put_url = server.data().get('metadata_put_url')
md = server.metadata_get()
metadata_url = md['os-collect-config']['request']['metadata_url']
self.assertNotEqual(metadata_url, metadata_put_url)
container_name = server.physical_resource_name()
object_name = server.data().get('metadata_object_name')
self.assertTrue(uuidutils.is_uuid_like(object_name))
test_path = '/v1/AUTH_test_tenant_id/%s/%s' % (
server.physical_resource_name(), object_name)
self.assertEqual(test_path, urlparse.urlparse(metadata_put_url).path)
self.assertEqual(test_path, urlparse.urlparse(metadata_url).path)
sc.put_object.assert_called_once_with(
container_name, object_name, jsonutils.dumps(md))
sc.head_container.return_value = {'x-container-object-count': '0'}
server._delete_temp_url()
sc.delete_object.assert_called_once_with(container_name, object_name)
sc.head_container.assert_called_once_with(container_name)
sc.delete_container.assert_called_once_with(container_name)
return metadata_url, server
def test_server_create_software_config_poll_temp_url(self):
metadata_url, server = (
self._server_create_software_config_poll_temp_url())
self.assertEqual({
'os-collect-config': {
'request': {
'metadata_url': metadata_url
},
'collectors': ['ec2', 'request', 'local']
},
'deployments': []
}, server.metadata_get())
def test_server_create_software_config_poll_temp_url_metadata(self):
md = {'os-collect-config': {'polling_interval': 10}}
metadata_url, server = (
self._server_create_software_config_poll_temp_url(md=md))
self.assertEqual({
'os-collect-config': {
'request': {
'metadata_url': metadata_url
},
'collectors': ['ec2', 'request', 'local'],
'polling_interval': 10
},
'deployments': []
}, server.metadata_get())
def test_delete_swift_service_removed(self):
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
return_server = self.fc.servers.list()[1]
stack_name = 'software_config_s'
(tmpl, stack) = self._setup_test_stack(stack_name)
props = tmpl.t['Resources']['WebServer']['Properties']
props['user_data_format'] = 'SOFTWARE_CONFIG'
props['software_config_transport'] = 'POLL_TEMP_URL'
self.server_props = props
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
sc = mock.Mock()
sc.head_account.return_value = {
'x-account-meta-temp-url-key': 'secrit'
}
sc.url = 'http://192.0.2.2'
self.patchobject(swift.SwiftClientPlugin, '_create',
return_value=sc)
self.patchobject(self.fc.servers, 'create',
return_value=return_server)
scheduler.TaskRunner(server.create)()
self.assertEqual((server.CREATE, server.COMPLETE), server.state)
self.patchobject(server.client_plugin(),
'does_endpoint_exist',
return_value=False)
side_effect = [server, fakes_nova.fake_exception()]
self.patchobject(self.fc.servers, 'get', side_effect=side_effect)
scheduler.TaskRunner(server.delete)()
self.assertEqual((server.DELETE, server.COMPLETE), server.state)
def _prepare_for_server_create(self, md=None):
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
return_server = self.fc.servers.list()[1]
stack_name = 'software_config_s'
(tmpl, stack) = self._setup_test_stack(stack_name)
props = tmpl.t['Resources']['WebServer']['Properties']
props['user_data_format'] = 'SOFTWARE_CONFIG'
props['software_config_transport'] = 'ZAQAR_MESSAGE'
if md is not None:
tmpl.t['Resources']['WebServer']['Metadata'] = md
self.server_props = props
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
self.patchobject(self.fc.servers, 'create',
return_value=return_server)
return server, stack
def _server_create_software_config_zaqar(self, md=None):
server, stack = self._prepare_for_server_create(md)
zcc = self.patchobject(zaqar.ZaqarClientPlugin, 'create_for_tenant')
zc = mock.Mock()
zcc.return_value = zc
queue = mock.Mock()
zc.queue.return_value = queue
scheduler.TaskRunner(server.create)()
metadata_queue_id = server.data().get('metadata_queue_id')
md = server.metadata_get()
queue_id = md['os-collect-config']['zaqar']['queue_id']
self.assertEqual(queue_id, metadata_queue_id)
zc.queue.assert_called_once_with(queue_id)
queue.post.assert_called_once_with(
{'body': server.metadata_get(), 'ttl': 3600})
zc.queue.reset_mock()
server._delete_queue()
zc.queue.assert_called_once_with(queue_id)
zc.queue(queue_id).delete.assert_called_once_with()
return queue_id, server
def test_server_create_software_config_zaqar(self):
queue_id, server = self._server_create_software_config_zaqar()
self.assertEqual({
'os-collect-config': {
'zaqar': {
'user_id': '1234',
'password': server.password,
'auth_url': 'http://server.test:5000/v2.0',
'project_id': '8888',
'queue_id': queue_id,
'region_name': 'RegionOne',
},
'collectors': ['ec2', 'zaqar', 'local']
},
'deployments': []
}, server.metadata_get())
def test_create_delete_no_zaqar_service(self):
zcc = self.patchobject(zaqar.ZaqarClientPlugin, 'create_for_tenant')
zcc.side_effect = ks_exceptions.EndpointNotFound
server, stack = self._prepare_for_server_create()
creator = scheduler.TaskRunner(server.create)
self.assertRaises(exception.ResourceFailure, creator)
self.assertEqual((server.CREATE, server.FAILED), server.state)
self.assertEqual({
'os-collect-config': {
'zaqar': {
'user_id': '1234',
'password': server.password,
'auth_url': 'http://server.test:5000/v2.0',
'project_id': '8888',
'queue_id': mock.ANY,
'region_name': 'RegionOne',
},
'collectors': ['ec2', 'zaqar', 'local']
},
'deployments': []
}, server.metadata_get())
scheduler.TaskRunner(server.delete)()
self.assertEqual((server.DELETE, server.COMPLETE), server.state)
def test_delete_zaqar_service_removed(self):
zcc = self.patchobject(zaqar.ZaqarClientPlugin, 'create_for_tenant')
zcc.return_value = mock.Mock()
server, stack = self._prepare_for_server_create()
scheduler.TaskRunner(server.create)()
self.assertEqual((server.CREATE, server.COMPLETE), server.state)
self.patchobject(server.client_plugin(),
'does_endpoint_exist',
return_value=False)
side_effect = [server, fakes_nova.fake_exception()]
self.patchobject(self.fc.servers, 'get', side_effect=side_effect)
scheduler.TaskRunner(server.delete)()
self.assertEqual((server.DELETE, server.COMPLETE), server.state)
def test_server_create_software_config_zaqar_metadata(self):
md = {'os-collect-config': {'polling_interval': 10}}
queue_id, server = self._server_create_software_config_zaqar(md=md)
self.assertEqual({
'os-collect-config': {
'zaqar': {
'user_id': '1234',
'password': server.password,
'auth_url': 'http://server.test:5000/v2.0',
'project_id': '8888',
'queue_id': queue_id,
'region_name': 'RegionOne',
},
'collectors': ['ec2', 'zaqar', 'local'],
'polling_interval': 10
},
'deployments': []
}, server.metadata_get())
def test_server_create_default_admin_pass(self):
return_server = self.fc.servers.list()[1]
return_server.adminPass = 'autogenerated'
stack_name = 'admin_pass_s'
(tmpl, stack) = self._setup_test_stack(stack_name)
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
mock_create = self.patchobject(self.fc.servers, 'create',
return_value=return_server)
scheduler.TaskRunner(server.create)()
_, kwargs = mock_create.call_args
self.assertIsNone(kwargs['admin_pass'])
self.assertEqual({}, kwargs['meta'])
def test_server_create_custom_admin_pass(self):
return_server = self.fc.servers.list()[1]
return_server.adminPass = 'foo'
stack_name = 'admin_pass_s'
(tmpl, stack) = self._setup_test_stack(stack_name)
tmpl.t['Resources']['WebServer']['Properties']['admin_pass'] = 'foo'
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
mock_create = self.patchobject(self.fc.servers, 'create',
return_value=return_server)
scheduler.TaskRunner(server.create)()
_, kwargs = mock_create.call_args
self.assertEqual('foo', kwargs['admin_pass'])
self.assertEqual({}, kwargs['meta'])
def test_server_create_with_stack_scheduler_hints(self):
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
return_server = self.fc.servers.list()[1]
return_server.id = '5678'
sh.cfg.CONF.set_override('stack_scheduler_hints', True)
# Unroll _create_test_server, to enable check
# for addition of heat ids (stack id, resource name)
stack_name = 'test_server_w_stack_sched_hints_s'
server_name = 'server_w_stack_sched_hints'
(t, stack) = self._get_test_template(stack_name, server_name)
self.patchobject(stack, 'path_in_stack',
return_value=[('parent', stack.name)])
resource_defns = t.resource_definitions(stack)
server = servers.Server(server_name,
resource_defns['WebServer'], stack)
self.patchobject(server, 'store_external_ports')
# server.uuid is only available once the resource has been added.
stack.add_resource(server)
self.assertIsNotNone(server.uuid)
mock_create = self.patchobject(self.fc.servers, 'create',
return_value=return_server)
shm = sh.SchedulerHintsMixin
scheduler_hints = {shm.HEAT_ROOT_STACK_ID: stack.root_stack_id(),
shm.HEAT_STACK_ID: stack.id,
shm.HEAT_STACK_NAME: stack.name,
shm.HEAT_PATH_IN_STACK: [','.join(['parent',
stack.name])],
shm.HEAT_RESOURCE_NAME: server.name,
shm.HEAT_RESOURCE_UUID: server.uuid}
scheduler.TaskRunner(server.create)()
_, kwargs = mock_create.call_args
self.assertEqual(scheduler_hints, kwargs['scheduler_hints'])
self.assertEqual({}, kwargs['meta'])
# this makes sure the auto increment worked on server creation
self.assertGreater(server.id, 0)
def test_check_maximum(self):
msg = 'test_check_maximum'
self.assertIsNone(servers.Server._check_maximum(1, 1, msg))
self.assertIsNone(servers.Server._check_maximum(1000, -1, msg))
error = self.assertRaises(exception.StackValidationFailed,
servers.Server._check_maximum,
2, 1, msg)
self.assertEqual(msg, str(error))
def test_server_validate(self):
stack_name = 'srv_val'
(tmpl, stack) = self._setup_test_stack(stack_name)
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('server_create_image',
resource_defns['WebServer'], stack)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
self.assertIsNone(server.validate())
def test_server_validate_with_bootable_vol(self):
stack_name = 'srv_val_bootvol'
(tmpl, stack) = self._setup_test_stack(stack_name)
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
self.stub_VolumeConstraint_validate()
# create a server with bootable volume
web_server = tmpl.t['Resources']['WebServer']
del web_server['Properties']['image']
def create_server(device_name):
web_server['Properties']['block_device_mapping'] = [{
"device_name": device_name,
"volume_id": "5d7e27da-6703-4f7e-9f94-1f67abef734c",
"delete_on_termination": False
}]
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('server_with_bootable_volume',
resource_defns['WebServer'], stack)
return server
server = create_server(u'vda')
self.assertIsNone(server.validate())
server = create_server('vda')
self.assertIsNone(server.validate())
server = create_server('vdb')
ex = self.assertRaises(exception.StackValidationFailed,
server.validate)
self.assertEqual('Neither image nor bootable volume is specified for '
'instance server_with_bootable_volume',
str(ex))
web_server['Properties']['image'] = ''
server = create_server('vdb')
self.assertIsNone(server.validate())
def test_server_validate_with_nova_keypair_resource(self):
stack_name = 'srv_val_test'
nova_keypair_template = '''
{
"AWSTemplateFormatVersion" : "2010-09-09",
"Description" : "WordPress",
"Resources" : {
"WebServer": {
"Type": "OS::Nova::Server",
"Properties": {
"image" : "F17-x86_64-gold",
"flavor" : "m1.large",
"key_name" : { "Ref": "SSHKey" },
"user_data" : "wordpress"
}
},
"SSHKey": {
"Type": "OS::Nova::KeyPair",
"Properties": {
"name": "my_key"
}
}
}
}
'''
t = template_format.parse(nova_keypair_template)
templ = template.Template(t)
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
stack = parser.Stack(utils.dummy_context(), stack_name, templ,
stack_id=uuidutils.generate_uuid())
resource_defns = templ.resource_definitions(stack)
server = servers.Server('server_validate_test',
resource_defns['WebServer'], stack)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
self.assertIsNone(server.validate())
def test_server_validate_with_invalid_ssh_key(self):
stack_name = 'srv_val_test'
(tmpl, stack) = self._setup_test_stack(stack_name)
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
web_server = tmpl['Resources']['WebServer']
# Make the ssh key have an invalid name
web_server['Properties']['key_name'] = 'test2'
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
error = self.assertRaises(exception.StackValidationFailed,
server.validate)
self.assertEqual(
"Property error: Resources.WebServer.Properties.key_name: "
"Error validating value 'test2': The Key (test2) could not "
"be found.", str(error))
def test_server_validate_software_config_invalid_meta(self):
stack_name = 'srv_val_test'
(tmpl, stack) = self._setup_test_stack(stack_name)
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
web_server = tmpl['Resources']['WebServer']
web_server['Properties']['user_data_format'] = 'SOFTWARE_CONFIG'
web_server['Metadata'] = {'deployments': 'notallowed'}
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('WebServer',
resource_defns['WebServer'], stack)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
error = self.assertRaises(exception.StackValidationFailed,
server.validate)
self.assertEqual(
"deployments key not allowed in resource metadata "
"with user_data_format of SOFTWARE_CONFIG", str(error))
def test_server_validate_with_networks(self):
stack_name = 'srv_net'
(tmpl, stack) = self._setup_test_stack(stack_name)
self.stub_KeypairConstraint_validate()
network_name = 'public'
# create a server with 'uuid' and 'network' properties
tmpl['Resources']['WebServer']['Properties']['networks'] = (
[{'uuid': 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa',
'network': network_name}])
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('server_validate_with_networks',
resource_defns['WebServer'], stack)
self.stub_NetworkConstraint_validate()
ex = self.assertRaises(exception.StackValidationFailed,
server.validate)
self.assertIn("Cannot define the following properties at "
"the same time: networks.network, networks.uuid",
str(ex))
def test_server_validate_with_network_empty_ref(self):
stack_name = 'srv_net'
(tmpl, stack) = self._setup_test_stack(stack_name)
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
tmpl['Resources']['WebServer']['Properties']['networks'] = (
[{'network': ''}])
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('server_validate_with_networks',
resource_defns['WebServer'], stack)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
self.patchobject(neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id')
self.assertIsNone(server.validate())
def test_server_validate_with_only_fixed_ip(self):
stack_name = 'srv_net'
(tmpl, stack) = self._setup_test_stack(stack_name)
# create a server with 'uuid' and 'network' properties
tmpl['Resources']['WebServer']['Properties']['networks'] = (
[{'fixed_ip': '10.0.0.99'}])
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('server_validate_with_networks',
resource_defns['WebServer'], stack)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
self.patchobject(neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id')
ex = self.assertRaises(exception.StackValidationFailed,
server.validate)
self.assertIn(_('One of the properties "network", "port", '
'"allocate_network" or "subnet" should be set '
'for the specified network of '
'server "%s".') % server.name,
str(ex))
def test_server_validate_with_network_floating_ip(self):
stack_name = 'srv_net_floating_ip'
(tmpl, stack) = self._setup_test_stack(stack_name)
# create a server with 'uuid' and 'network' properties
tmpl['Resources']['WebServer']['Properties']['networks'] = (
[{'floating_ip': '172.24.4.14',
'network': '6b1688bb-18a0-4754-ab05-19daaedc5871'}])
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('server_validate_net_floating_ip',
resource_defns['WebServer'], stack)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
self.patchobject(neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id')
ex = self.assertRaises(exception.StackValidationFailed,
server.validate)
self.assertIn(_('Property "floating_ip" is not supported if '
'only "network" is specified, because the '
'corresponding port can not be retrieved.'),
str(ex))
def test_server_validate_with_networks_str_net(self):
stack_name = 'srv_networks_str_nets'
(tmpl, stack) = self._setup_test_stack(stack_name)
# create a server with 'uuid' and 'network' properties
tmpl['Resources']['WebServer']['Properties']['networks'] = (
[{'network': '6b1688bb-18a0-4754-ab05-19daaedc5871',
'allocate_network': 'auto'}])
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('server_validate_net_list_str',
resource_defns['WebServer'], stack)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
self.patchobject(neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id')
ex = self.assertRaises(exception.StackValidationFailed,
server.validate)
self.assertIn(_('Can not specify "allocate_network" with '
'other keys of networks at the same time.'),
str(ex))
def test_server_validate_port_fixed_ip(self):
stack_name = 'port_with_fixed_ip'
(tmpl, stack) = self._setup_test_stack(stack_name,
test_templ=with_port_template)
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('validate_port_reference_fixed_ip',
resource_defns['server'], stack)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
error = self.assertRaises(exception.ResourcePropertyConflict,
server.validate)
self.assertEqual("Cannot define the following properties at the same "
"time: networks/fixed_ip, networks/port.",
str(error))
# test if the 'port' doesn't reference with non-created resource
tmpl['Resources']['server']['Properties']['networks'] = (
[{'port': 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa',
'fixed_ip': '10.0.0.99'}])
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('with_port_fixed_ip',
resource_defns['server'], stack)
self.patchobject(neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id')
error = self.assertRaises(exception.ResourcePropertyConflict,
server.validate)
self.assertEqual("Cannot define the following properties at the same "
"time: networks/fixed_ip, networks/port.",
str(error))
def test_server_validate_with_uuid_fixed_ip(self):
stack_name = 'srv_net'
(tmpl, stack) = self._setup_test_stack(stack_name)
tmpl['Resources']['WebServer']['Properties']['networks'] = (
[{'uuid': 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa',
'fixed_ip': '10.0.0.99'}])
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('server_validate_with_networks',
resource_defns['WebServer'], stack)
self.patchobject(neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id')
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
self.assertIsNone(server.validate())
def test_server_validate_with_network_fixed_ip(self):
stack_name = 'srv_net'
(tmpl, stack) = self._setup_test_stack(stack_name)
tmpl['Resources']['WebServer']['Properties']['networks'] = (
[{'network': 'public',
'fixed_ip': '10.0.0.99'}])
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('server_validate_with_networks',
resource_defns['WebServer'], stack)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
self.patchobject(neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id')
self.assertIsNone(server.validate())
def test_server_validate_net_security_groups(self):
# Test that if network 'ports' are assigned security groups are
# not, because they'll be ignored
stack_name = 'srv_net_secgroups'
(tmpl, stack) = self._setup_test_stack(stack_name)
tmpl['Resources']['WebServer']['Properties']['networks'] = [
{'port': ''}]
tmpl['Resources']['WebServer']['Properties'][
'security_groups'] = ['my_security_group']
self.patchobject(nova.NovaClientPlugin, 'client',
return_value=self.fc)
resource_defns = tmpl.resource_definitions(stack)
server = servers.Server('server_validate_net_security_groups',
resource_defns['WebServer'], stack)
self.patchobject(glance.GlanceClientPlugin, 'get_image',
return_value=self.mock_image)
self.patchobject(nova.NovaClientPlugin, 'get_flavor',
return_value=self.mock_flavor)
self.patchobject(neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id')
error = self.assertRaises(exception.ResourcePropertyConflict,
server.validate)
self.assertEqual("Cannot define the following properties at the same "
"time: security_groups, networks/port.",
str(error))
def test_server_delete(self):
return_server = self.fc.servers.list()[1]
server = self._create_test_server(return_server,
'create_delete')
server.resource_id = '1234'
# this makes sure the auto increment worked on server creation
self.assertGreater(server.id, 0)
side_effect = [server, fakes_nova.fake_exception()]
self.patchobject(self.fc.servers, 'get', side_effect=side_effect)
scheduler.TaskRunner(server.delete)()
self.assertEqual((server.DELETE, server.COMPLETE), server.state)
def test_server_delete_notfound(self):
return_server = self.fc.servers.list()[1]
server = self._create_test_server(return_server,
'create_delete2')
server.resource_id = '1234'
# this makes sure the auto increment worked on server creation
self.assertGreater(server.id, 0)
self.patchobject(self.fc.client, 'delete_servers_1234',
side_effect=fakes_nova.fake_exception())
scheduler.TaskRunner(server.delete)()
self.assertEqual((server.DELETE, server.COMPLETE), server.state)
def test_server_delete_error(self):
return_server = self.fc.servers.list()[1]
server = self._create_test_server(return_server,
'create_delete')
server.resource_id = '1234'
# this makes sure the auto increment worked on server creation
self.assertGreater(server.id, 0)
def make_error(*args):
return_server.status = "ERROR"
return return_server
self.patchobject(self.fc.servers, 'get',
side_effect=[return_server, return_server,
make_error()])
resf = self.assertRaises(exception.ResourceFailure,
scheduler.TaskRunner(server.delete))
self.assertIn("Server %s delete failed" % return_server.name,
str(resf))
def test_server_delete_error_task_in_progress(self):
# test server in 'ERROR', but task state in nova is 'deleting'
return_server = self.fc.servers.list()[1]
server = self._create_test_server(return_server,
'create_delete')
server.resource_id = '1234'
def make_error(*args):
return_server.status = "ERROR"
setattr(return_server, 'OS-EXT-STS:task_state', 'deleting')
return return_server
def make_error_done(*args):
return_server.status = "ERROR"
setattr(return_server, 'OS-EXT-STS:task_state', None)
return return_server
self.patchobject(self.fc.servers, 'get',
side_effect=[make_error(),
make_error_done()])
resf = self.assertRaises(exception.ResourceFailure,
scheduler.TaskRunner(server.delete))
self.assertIn("Server %s delete failed" % return_server.name,
str(resf))
def test_server_soft_delete(self):
return_server = self.fc.servers.list()[1]
server = self._create_test_server(return_server,
'create_delete')
server.resource_id = '1234'
# this makes sure the auto increment worked on server creation
self.assertGreater(server.id, 0)
self.patchobject(self.fc.servers, 'get',
side_effect=ServerStatus(return_server,
[return_server.status,
return_server.status,
"SOFT_DELETED",
"DELETED"]))
scheduler.TaskRunner(server.delete)()
self.assertEqual((server.DELETE, server.COMPLETE), server.state)
def test_server_update_metadata(self):
return_server = self.fc.servers.list()[1]
server = self._create_test_server(return_server,
'md_update')
ud_tmpl = self._get_test_template('update_stack')[0]
ud_tmpl.t['Resources']['WebServer']['Metadata'] = {'test': 123}
resource_defns = ud_tmpl.resource_definitions(server.stack)
scheduler.TaskRunner(server.update, resource_defns['WebServer'])()
self.assertEqual({'test': 123}, server.metadata_get())
ud_tmpl.t['Resources']['WebServer']['Metadata'] = {'test': 456}
server.t = ud_tmpl.resource_definitions(server.stack)['WebServer']
self.assertEqual({'test': 123}, server.metadata_get())
server.metadata_update()
self.assertEqual({'test': 456}, server.metadata_get())
@mock.patch.object(heat_plugin.HeatClientPlugin, 'url_for')
def test_server_update_metadata_software_config(self, fake_url):
fake_url.return_value = 'http://ip:8000/v1'
server, ud_tmpl = self._server_create_software_config(
stack_name='update_meta_sc', ret_tmpl=True)
expected_md = {
'os-collect-config': {
'cfn': {
'access_key_id': '4567',
'metadata_url': 'http://ip:8000/v1/',
'path': 'WebServer.Metadata',
'secret_access_key': '8901',
'stack_name': 'update_meta_sc'
},
'collectors': ['ec2', 'cfn', 'local']
},
'deployments': []}
self.assertEqual(expected_md, server.metadata_get())
ud_tmpl.t['Resources']['WebServer']['Metadata'] = {'test': 123}
resource_defns = ud_tmpl.resource_definitions(server.stack)
scheduler.TaskRunner(server.update, resource_defns['WebServer'])()
expected_md.update({'test': 123})
self.assertEqual(expected_md, server.metadata_get())
server.metadata_update()
self.assertEqual(expected_md, server.metadata_get())
@mock.patch.object(heat_plugin.HeatClientPlugin, 'url_for')
def test_server_update_metadata_software_config_merge(self, fake_url):
md = {'os-collect-config': {'polling_interval': 10}}
fake_url.return_value = 'http://ip/heat-api-cfn/v1'
server, ud_tmpl = self._server_create_software_config(
stack_name='update_meta_sc', ret_tmpl=True,
md=md)
expected_md = {
'os-collect-config': {
'cfn': {
'access_key_id': '4567',
'metadata_url': 'http://ip/heat-api-cfn/v1/',
'path': 'WebServer.Metadata',
'secret_access_key': '8901',
'stack_name': 'update_meta_sc'
},
'collectors': ['ec2', 'cfn', 'local'],
'polling_interval': 10
},
'deployments': []}
self.assertEqual(expected_md, server.metadata_get())
ud_tmpl.t['Resources']['WebServer']['Metadata'] = {'test': 123}
resource_defns = ud_tmpl.resource_definitions(server.stack)
scheduler.TaskRunner(server.update, resource_defns['WebServer'])()
expected_md.update({'test': 123})
self.assertEqual(expected_md, server.metadata_get())
server.metadata_update()
self.assertEqual(expected_md, server.metadata_get())
@mock.patch.object(heat_plugin.HeatClientPlugin, 'url_for')
def test_server_update_software_config_transport(self, fake_url):
md = {'os-collect-config': {'polling_interval': 10}}
fake_url.return_value = 'http://ip/heat-api-cfn/v1'
server = self._server_create_software_config(
stack_name='update_meta_sc', md=md)
expected_md = {
'os-collect-config': {
'cfn': {
'access_key_id': '4567',
'metadata_url': 'http://ip/heat-api-cfn/v1/',
'path': 'WebServer.Metadata',
'secret_access_key': '8901',
'stack_name': 'update_meta_sc'
},
'collectors': ['ec2', 'cfn', 'local'],
'polling_interval': 10
},
'deployments': []}
self.assertEqual(expected_md, server.metadata_get())
sc = mock.Mock()
sc.head_account.return_value = {
'x-account-meta-temp-url-key': 'secrit'
}
sc.url = 'http://192.0.2.2'
self.patchobject(swift.SwiftClientPlugin, '_create',
return_value=sc)
update_props = self.server_props.copy()
update_props['software_config_transport'] = 'POLL_TEMP_URL'
update_template = server.t.freeze(properties=update_props)
self.rpc_client = mock.MagicMock()
server._rpc_client = self.rpc_client
self.rpc_client.create_software_config.return_value = None
scheduler.TaskRunner(server.update, update_template)()
self.assertEqual((server.UPDATE, server.COMPLETE), server.state)
md = server.metadata_get()
metadata_url = md['os-collect-config']['request']['metadata_url']
self.assertTrue(metadata_url.startswith(
'http://192.0.2.2/v1/AUTH_test_tenant_id/'))
expected_md = {
'os-collect-config': {
'cfn': {
'access_key_id': None,
'metadata_url': None,
'path': None,
'secret_access_key': None,
'stack_name': None,
},
'request': {
'metadata_url': 'the_url',
},
'collectors': ['ec2', 'request', 'local'],
'polling_interval': 10
},
'deployments': []}
md['os-collect-config']['request']['metadata_url'] = 'the_url'
self.assertEqual(expected_md, server.metadata_get())
def test_update_transport_heat_to_zaqar(self):
stack, server = self._server_create_software_config_poll_heat()
password = server.password
self.assertEqual({
'os-collect-config': {
'heat': {
'auth_url': 'http://server.test:5000/v2.0',
'password': password,
'project_id': '8888',
'region_name': 'RegionOne',
'resource_name': 'WebServer',
'stack_id': 'software_config_s/%s' % stack.id,
'user_id': '1234'
},
'collectors': ['ec2', 'heat', 'local'],
},
'deployments': []
}, server.metadata_get())
update_props = self.server_props.copy()
update_props['software_config_transport'] = 'ZAQAR_MESSAGE'
update_template = server.t.freeze(properties=update_props)
zcc = self.patchobject(zaqar.ZaqarClientPlugin, 'create_for_tenant')
zc = mock.Mock()
zcc.return_value = zc
queue = mock.Mock()
zc.queue.return_value = queue
self.rpc_client = mock.MagicMock()
server._rpc_client = self.rpc_client
self.rpc_client.create_software_config.return_value = None
scheduler.TaskRunner(server.update, update_template)()
self.assertEqual((server.UPDATE, server.COMPLETE), server.state)
password_1 = server.password
self.assertEqual(password, password_1)
self.assertEqual({
'os-collect-config': {
'zaqar': {
'user_id': '1234',
'password': password_1,
'auth_url': 'http://server.test:5000/v2.0',
'project_id': '8888',
'queue_id': server.data().get('metadata_queue_id'),
'region_name': 'RegionOne',
},
'heat': {
'auth_url': None,
'password': None,
'project_id': None,
'region_name': None,
'resource_name': None,
'stack_id': None,
'user_id': None
},
'collectors': ['ec2', 'zaqar', 'local']
},
'deployments': []
}, server.metadata_get())
def test_server_update_nova_metadata(self):
return_server = self.fc.servers.list()[1]
server = self._create_test_server(return_server,
'md_update')
new_meta = {'test': 123}
self.patchobject(self.fc.servers, 'get',
return_value=return_server)
set_meta_mock = self.patchobject(self.fc.servers, 'set_meta')
update_props = self.server_props.copy()
update_props['metadata'] = new_meta
update_template = server.t.freeze(properties=update_props)
scheduler.TaskRunner(server.update, update_template)()
self.assertEqual((server.UPDATE, server.COMPLETE), server.state)
set_meta_mock.assert_called_with(
return_server, server.client_plugin().meta_serialize(new_meta))
def test_server_update_nova_metadata_complex(self):
"""Test that complex metadata values are correctly serialized to JSON.
Test that complex metadata values are correctly serialized to JSON when
sent to Nova.
"""
return_server = self.fc.servers.list()[1]
server = self._create_test_server(return_server,
'md_update')
self.patchobject(self.fc.servers, 'get',
return_value=return_server)
new_meta = {'test': {'testkey': 'testvalue'}}
set_meta_mock = self.patchobject(self.fc.servers, 'set_meta')
# If we're going to call set_meta() directly we
# need to handle the serialization ourselves.
update_props = self.server_props.copy()
update_props['metadata'] = new_meta
update_template = server.t.freeze(properties=update_props)
scheduler.TaskRunner(server.update, update_template)()
self.assertEqual((server.UPDATE, server.COMPLETE), server.state)
set_meta_mock.assert_called_with(
return_server, server.client_plugin().meta_serialize(new_meta))
def test_server_update_nova_metadata_with_delete(self):
return_server = self.fc.servers.list()[1]
server = self._create_test_server(return_server,
'md_update')
# part one, add some metadata
new_meta = {'test': '123', 'this': 'that'}
self.patchobject(self.fc.servers, 'get',
return_value=return_server)
set_meta_mock = self.patchobject(self.fc.servers, 'set_meta')
update_props = self.server_props.copy()
update_props['metadata'] = new_meta
update_template = server.t.freeze(properties=update_props)
scheduler.TaskRunner(server.update, update_template)()
self.assertEqual((server.UPDATE, server.COMPLETE), server.state)
set_meta_mock.assert_called_with(
return_server, server.client_plugin().meta_serialize(new_meta))
# part two change the metadata (test removing the old key)
new_meta = {'new_key': 'yeah'}
# new fake with the correct metadata
server.resource_id = '56789'
new_return_server = self.fc.servers.list()[5]
self.patchobject(self.fc.servers, 'get',
return_value=new_return_server)
del_meta_mock = self.patchobject(self.fc.servers, 'delete_meta')
update_props = self.server_props.copy()
update_props['metadata'] = new_meta
update_template = server.t.freeze(properties=update_props)
scheduler.TaskRunner(server.update, update_template)()
self.assertEqual((server.UPDATE, server.COMPLETE), server.state)
del_meta_mock.assert_called_with(new_return_server,
['test', 'this'])
set_meta_mock.assert_called_with(
new_return_server, server.client_plugin().meta_serialize(new_meta))
def test_server_update_server_name(self):
"""Server.handle_update supports changing the name."""
return_server = self.fc.servers.list()[1]
return_server.id = '5678'
server = self._create_test_server(return_server,
'srv_update')
new_name = 'new_name'
update_props = self.server_props.copy()
update_props['name'] = new_name
update_template = server.t.freeze(properties=update_props)
self.patchobject(self.fc.servers, 'get',
return_value=return_server)
self.patchobject(return_server, 'update')
return_server.update(new_name).AndReturn(None)
scheduler.TaskRunner(server.update, update_template)()
self.assertEqual((server.UPDATE, server.COMPLETE), server.state)
def test_server_update_server_admin_password(self):
"""Server.handle_update supports changing the admin password."""
return_server = self.fc.servers.list()[1]
return_server.id = '5678'
server = self._create_test_server(return_server,
'change_password')
new_password = 'new_password'
update_props = self.server_props.copy()
update_props['admin_pass'] = new_password
update_template = server.t.freeze(properties=update_props)
self.patchobject(self.fc.servers, 'get', return_value=return_server)
self.patchobject(return_server, 'change_password')
scheduler.TaskRunner(server.update, update_template)()
self.assertEqual((server.UPDATE, server.COMPLETE), server.state)
return_server.change_password.assert_called_once_with(new_password)
self.assertEqual(1, return_server.change_password.call_count)
def test_server_get_live_state(self):
return_server = self.fc.servers.list()[1]
return_server.id = '5678'
self.patchobject(nova.NovaClientPlugin, 'is_version_supported',
return_value=False)
server = self._create_test_server(return_server,
'get_live_state_stack')
server.properties.data['networks'] = [{'network': 'public_id',
'fixed_ip': '5.6.9.8'}]
public_net = dict(id='public_id', name='public')
private_net = dict(id='private_id', name='private')
iface0 = create_fake_iface(port='aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa',
net='public',
ip='5.6.9.8',
mac='fa:16:3e:8c:33:aa')
port0 = dict(id=iface0.port_id,
network_id=iface0.net_id,
mac_address=iface0.mac_addr,
fixed_ips=iface0.fixed_ips)
iface1 = create_fake_iface(port='bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
net='public',
ip='4.5.6.7',
mac='fa:16:3e:8c:22:aa')
port1 = dict(id=iface1.port_id,
network_id=iface1.net_id,
mac_address=iface1.mac_addr,
fixed_ips=iface1.fixed_ips)
iface2 = create_fake_iface(port='cccccccc-cccc-cccc-cccc-cccccccccccc',
net='private',
ip='10.13.12.13',
mac='fa:16:3e:8c:44:cc')
port2 = dict(id=iface2.port_id,
network_id=iface2.net_id,
mac_address=iface2.mac_addr,
fixed_ips=iface2.fixed_ips)
self.patchobject(return_server, 'interface_list',
return_value=[iface0, iface1, iface2])
self.patchobject(neutronclient.Client, 'list_ports',
return_value={'ports': [port0, port1, port2]})
self.patchobject(neutronclient.Client, 'list_networks',
side_effect=[{'networks': [public_net]},
{'networks': [public_net]},
{'networks': [private_net]}])
self.patchobject(neutronclient.Client, 'list_floatingips',
return_value={'floatingips': []})
self.patchobject(neutron.NeutronClientPlugin,
'find_resourceid_by_name_or_id',
side_effect=['public_id',
'private_id'])
reality = server.get_live_state(server.properties.data)
expected = {'flavor': '1',
'image': '2',
'name': 'sample-server2',
'networks': [
{'fixed_ip': '4.5.6.7',
'network': 'public',
'port': 'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb'},
{'fixed_ip': '5.6.9.8',
'network': 'public',
'port': None},
{'fixed_ip': '10.13.12.13',
'network': 'private',
'port': 'cccccccc-cccc-cccc-cccc-cccccccccccc'}],
'metadata': {}}
self.assertEqual(set(expected.keys()), set(reality.keys()))
expected_nets = expected.pop('networks')
reality_nets = reality.pop('networks')
for net in reality_nets:
for exp_net in expected_nets:
if net == exp_net:
for key in net:
self.assertEqual(exp_net[key], net[key])
break
for key in reality.keys():
self.assertEqual(expected[key], reality[key])
def test_server_update_server_flavor(self):
"""Tests update server changing the flavor.
Server.handle_update supports changing the flavor, and makes
the change making a resize API call against Nova.
"""
return_server = self.fc.servers.list()[1]
return_server.id = '1234'
server = self._create_test_server(return_server,
'srv_update')
update_props = self.server_props.copy()
update_props['flavor'] = 'm1.small'
update_template = server.t.freeze(properties=update_props)
self.patchobject(self.fc.servers, 'get',
side_effect=ServerStatus(return_server,
['ACTIVE',
'RESIZE',
'VERIFY_RESIZE',
'VERIFY_RESIZE',
'ACTIVE']))
mock_post = self.patchobject(self.fc.client,
'post_servers_1234_action',
return_value=(202, None))