1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Licensed under the Apache License, Version 2.0 (the "License"); you may
4
# not use this file except in compliance with the License. You may obtain
5
# a copy of the License at
7
# http://www.apache.org/licenses/LICENSE-2.0
9
# Unless required by applicable law or agreed to in writing, software
10
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
# License for the specific language governing permissions and limitations
19
from heat.engine import environment
20
from heat.tests.v1_1 import fakes
21
from heat.common import exception
22
from heat.common import template_format
23
from heat.engine import parser
24
from heat.engine import resource
25
from heat.engine import scheduler
26
from heat.engine.resources import server as servers
27
from heat.openstack.common import uuidutils
28
from heat.tests.common import HeatTestCase
29
from heat.tests import utils
34
"AWSTemplateFormatVersion" : "2010-09-09",
35
"Description" : "WordPress",
38
"Description" : "key_name",
45
"Type": "OS::Nova::Server",
47
"image" : "F17-x86_64-gold",
48
"flavor" : "m1.large",
50
"user_data" : "wordpress"
58
class ServersTest(HeatTestCase):
60
super(ServersTest, self).setUp()
61
self.fc = fakes.FakeClient()
62
utils.setup_dummy_db()
64
def _setup_test_stack(self, stack_name):
65
t = template_format.parse(wp_template)
66
template = parser.Template(t)
67
stack = parser.Stack(utils.dummy_context(), stack_name, template,
68
environment.Environment({'key_name': 'test'}),
69
stack_id=uuidutils.generate_uuid())
72
def _setup_test_server(self, return_server, name, image_id=None):
73
stack_name = '%s_stack' % name
74
(t, stack) = self._setup_test_stack(stack_name)
76
t['Resources']['WebServer']['Properties']['image'] = \
77
image_id or 'CentOS 5.2'
78
t['Resources']['WebServer']['Properties']['flavor'] = \
80
server = servers.Server('%s_name' % name,
81
t['Resources']['WebServer'], stack)
83
self.m.StubOutWithMock(server, 'nova')
84
server.nova().MultipleTimes().AndReturn(self.fc)
86
server.t = server.stack.resolve_runtime_data(server.t)
88
# need to resolve the template functions
89
#server_userdata = nova_utils.build_userdata(
91
# server.t['Properties']['user_data'])
92
#server.mime_string = server_userdata
93
self.m.StubOutWithMock(self.fc.servers, 'create')
94
self.fc.servers.create(
95
image=1, flavor=1, key_name='test',
96
name=utils.PhysName(stack_name, server.name),
98
userdata=mox.IgnoreArg(), scheduler_hints=None,
99
meta=None, nics=None, availability_zone=None,
100
block_device_mapping=None, config_drive=None,
101
disk_config=None, reservation_id=None).AndReturn(
106
def _create_test_server(self, return_server, name):
107
server = self._setup_test_server(return_server, name)
109
scheduler.TaskRunner(server.create)()
112
def test_server_create(self):
113
return_server = self.fc.servers.list()[1]
114
server = self._create_test_server(return_server,
115
'test_server_create')
116
# this makes sure the auto increment worked on server creation
117
self.assertTrue(server.id > 0)
119
public_ip = return_server.networks['public'][0]
121
server.FnGetAtt('addresses')['public'][0]['addr'], public_ip)
123
server.FnGetAtt('networks')['public'][0], public_ip)
125
server.FnGetAtt('first_public_address'), public_ip)
127
private_ip = return_server.networks['private'][0]
129
server.FnGetAtt('addresses')['private'][0]['addr'], private_ip)
131
server.FnGetAtt('networks')['private'][0], private_ip)
133
server.FnGetAtt('first_private_address'), private_ip)
135
self.assertEqual(return_server._info, server.FnGetAtt('show'))
136
self.assertEqual('sample-server2', server.FnGetAtt('instance_name'))
137
self.assertEqual('192.0.2.0', server.FnGetAtt('accessIPv4'))
138
self.assertEqual('::babe:4317:0A83', server.FnGetAtt('accessIPv6'))
141
def test_server_create_with_image_id(self):
142
return_server = self.fc.servers.list()[1]
143
server = self._setup_test_server(return_server,
144
'test_server_create_image_id',
146
self.m.StubOutWithMock(uuidutils, "is_uuid_like")
147
uuidutils.is_uuid_like('1').AndReturn(True)
150
scheduler.TaskRunner(server.create)()
152
# this makes sure the auto increment worked on server creation
153
self.assertTrue(server.id > 0)
155
public_ip = return_server.networks['public'][0]
157
server.FnGetAtt('addresses')['public'][0]['addr'], public_ip)
159
server.FnGetAtt('networks')['public'][0], public_ip)
161
server.FnGetAtt('first_public_address'), public_ip)
163
private_ip = return_server.networks['private'][0]
165
server.FnGetAtt('addresses')['private'][0]['addr'], private_ip)
167
server.FnGetAtt('networks')['private'][0], private_ip)
169
server.FnGetAtt('first_private_address'), private_ip)
173
def test_server_create_image_name_err(self):
174
stack_name = 'test_server_create_image_name_err_stack'
175
(t, stack) = self._setup_test_stack(stack_name)
177
# create an server with non exist image name
178
t['Resources']['WebServer']['Properties']['image'] = 'Slackware'
179
server = servers.Server('server_create_image_err',
180
t['Resources']['WebServer'], stack)
182
self.m.StubOutWithMock(server, 'nova')
183
server.nova().MultipleTimes().AndReturn(self.fc)
186
self.assertRaises(exception.ImageNotFound, server.handle_create)
190
def test_server_create_duplicate_image_name_err(self):
191
stack_name = 'test_server_create_image_name_err_stack'
192
(t, stack) = self._setup_test_stack(stack_name)
194
# create an server with a non unique image name
195
t['Resources']['WebServer']['Properties']['image'] = 'CentOS 5.2'
196
server = servers.Server('server_create_image_err',
197
t['Resources']['WebServer'], stack)
199
self.m.StubOutWithMock(server, 'nova')
200
server.nova().MultipleTimes().AndReturn(self.fc)
201
self.m.StubOutWithMock(self.fc.client, "get_images_detail")
202
self.fc.client.get_images_detail().AndReturn((
203
200, {'images': [{'id': 1, 'name': 'CentOS 5.2'},
204
{'id': 4, 'name': 'CentOS 5.2'}]}))
207
self.assertRaises(exception.NoUniqueImageFound, server.handle_create)
211
def test_server_create_image_id_err(self):
212
stack_name = 'test_server_create_image_id_err_stack'
213
(t, stack) = self._setup_test_stack(stack_name)
215
# create an server with non exist image Id
216
t['Resources']['WebServer']['Properties']['image'] = '1'
217
server = servers.Server('server_create_image_err',
218
t['Resources']['WebServer'], stack)
220
self.m.StubOutWithMock(server, 'nova')
221
server.nova().MultipleTimes().AndReturn(self.fc)
222
self.m.StubOutWithMock(uuidutils, "is_uuid_like")
223
uuidutils.is_uuid_like('1').AndReturn(True)
224
self.m.StubOutWithMock(self.fc.client, "get_images_1")
225
self.fc.client.get_images_1().AndRaise(
226
servers.clients.novaclient.exceptions.NotFound(404))
229
self.assertRaises(exception.ImageNotFound, server.handle_create)
233
def test_server_create_unexpected_status(self):
234
return_server = self.fc.servers.list()[1]
235
server = self._create_test_server(return_server,
236
'test_server_create')
237
return_server.get = lambda: None
238
return_server.status = 'BOGUS'
239
self.assertRaises(exception.Error,
240
server.check_create_complete,
243
def test_server_create_error_status(self):
244
return_server = self.fc.servers.list()[1]
245
server = self._create_test_server(return_server,
246
'test_server_create')
247
return_server.status = 'ERROR'
248
return_server.fault = {
249
'message': 'NoValidHost',
251
'created': '2013-08-14T03:12:10Z'
253
self.m.StubOutWithMock(return_server, 'get')
257
self.assertRaises(exception.Error,
258
server.check_create_complete,
263
def test_server_validate(self):
264
stack_name = 'test_server_validate_stack'
265
(t, stack) = self._setup_test_stack(stack_name)
267
# create an server with non exist image Id
268
t['Resources']['WebServer']['Properties']['image'] = '1'
269
server = servers.Server('server_create_image_err',
270
t['Resources']['WebServer'], stack)
272
self.m.StubOutWithMock(server, 'nova')
273
server.nova().MultipleTimes().AndReturn(self.fc)
275
self.m.StubOutWithMock(uuidutils, "is_uuid_like")
276
uuidutils.is_uuid_like('1').AndReturn(True)
279
self.assertEqual(server.validate(), None)
283
def test_server_validate_delete_policy(self):
284
stack_name = 'test_server_validate_stack'
285
(t, stack) = self._setup_test_stack(stack_name)
287
# create an server with non exist image Id
288
t['Resources']['WebServer']['DeletionPolicy'] = 'SelfDestruct'
289
server = servers.Server('server_create_image_err',
290
t['Resources']['WebServer'], stack)
294
ex = self.assertRaises(exception.StackValidationFailed,
296
self.assertEqual('Invalid DeletionPolicy SelfDestruct',
301
def test_server_delete(self):
302
return_server = self.fc.servers.list()[1]
303
server = self._create_test_server(return_server,
304
'test_server_create_delete')
305
server.resource_id = 1234
307
# this makes sure the auto increment worked on server creation
308
self.assertTrue(server.id > 0)
310
server_get = self.fc.client.get_servers_1234()
311
self.m.StubOutWithMock(self.fc.client, 'get_servers_1234')
312
get = self.fc.client.get_servers_1234
313
get().AndReturn(server_get)
314
get().AndRaise(servers.clients.novaclient.exceptions.NotFound(404))
318
scheduler.TaskRunner(server.delete)()
319
self.assertTrue(server.resource_id is None)
320
self.assertEqual(server.state, (server.DELETE, server.COMPLETE))
323
def test_server_delete_notfound(self):
324
return_server = self.fc.servers.list()[1]
325
server = self._create_test_server(return_server,
326
'test_server_create_delete')
327
server.resource_id = 1234
329
# this makes sure the auto increment worked on server creation
330
self.assertTrue(server.id > 0)
332
self.m.StubOutWithMock(self.fc.client, 'get_servers_1234')
333
get = self.fc.client.get_servers_1234
334
get().AndRaise(servers.clients.novaclient.exceptions.NotFound(404))
337
scheduler.TaskRunner(server.delete)()
338
self.assertTrue(server.resource_id is None)
339
self.assertEqual(server.state, (server.DELETE, server.COMPLETE))
342
server.state_set(server.CREATE, server.COMPLETE, 'to delete again')
343
scheduler.TaskRunner(server.delete)()
344
self.assertEqual(server.state, (server.DELETE, server.COMPLETE))
347
def test_server_update_metadata(self):
348
return_server = self.fc.servers.list()[1]
349
server = self._create_test_server(return_server,
350
'test_server_update')
352
update_template = copy.deepcopy(server.t)
353
update_template['Metadata'] = {'test': 123}
354
scheduler.TaskRunner(server.update, update_template)()
355
self.assertEqual(server.metadata, {'test': 123})
357
server.t['Metadata'] = {'test': 456}
358
server.metadata_update()
359
self.assertEqual(server.metadata, {'test': 456})
361
def test_server_update_server_flavor(self):
363
Server.handle_update supports changing the flavor, and makes
364
the change making a resize API call against Nova.
366
return_server = self.fc.servers.list()[1]
367
return_server.id = 1234
368
server = self._create_test_server(return_server,
369
'test_server_update')
371
update_template = copy.deepcopy(server.t)
372
update_template['Properties']['flavor'] = 'm1.small'
374
self.m.StubOutWithMock(self.fc.servers, 'get')
375
self.fc.servers.get(1234).AndReturn(return_server)
377
def activate_status(server):
378
server.status = 'VERIFY_RESIZE'
379
return_server.get = activate_status.__get__(return_server)
381
self.m.StubOutWithMock(self.fc.client, 'post_servers_1234_action')
382
self.fc.client.post_servers_1234_action(
383
body={'resize': {'flavorRef': 2}}).AndReturn((202, None))
384
self.fc.client.post_servers_1234_action(
385
body={'confirmResize': None}).AndReturn((202, None))
388
scheduler.TaskRunner(server.update, update_template)()
389
self.assertEqual(server.state, (server.UPDATE, server.COMPLETE))
392
def test_server_update_server_flavor_failed(self):
394
If the status after a resize is not VERIFY_RESIZE, it means the resize
395
call failed, so we raise an explicit error.
397
return_server = self.fc.servers.list()[1]
398
return_server.id = 1234
399
server = self._create_test_server(return_server,
400
'test_server_update')
402
update_template = copy.deepcopy(server.t)
403
update_template['Properties']['flavor'] = 'm1.small'
405
self.m.StubOutWithMock(self.fc.servers, 'get')
406
self.fc.servers.get(1234).AndReturn(return_server)
408
def activate_status(server):
409
server.status = 'ACTIVE'
410
return_server.get = activate_status.__get__(return_server)
412
self.m.StubOutWithMock(self.fc.client, 'post_servers_1234_action')
413
self.fc.client.post_servers_1234_action(
414
body={'resize': {'flavorRef': 2}}).AndReturn((202, None))
417
updater = scheduler.TaskRunner(server.update, update_template)
418
error = self.assertRaises(exception.ResourceFailure, updater)
420
"Error: Resizing to 'm1.small' failed, status 'ACTIVE'",
422
self.assertEqual(server.state, (server.UPDATE, server.FAILED))
425
def test_server_update_server_flavor_replace(self):
426
stack_name = 'test_server_update_flavor_replace'
427
(t, stack) = self._setup_test_stack(stack_name)
429
t['Resources']['WebServer']['Properties'][
430
'flavor_update_policy'] = 'REPLACE'
431
server = servers.Server('server_server_update_flavor_replace',
432
t['Resources']['WebServer'], stack)
434
update_template = copy.deepcopy(server.t)
435
update_template['Properties']['flavor'] = 'm1.smigish'
436
updater = scheduler.TaskRunner(server.update, update_template)
437
self.assertRaises(resource.UpdateReplace, updater)
439
def test_server_update_server_flavor_policy_update(self):
440
stack_name = 'test_server_update_flavor_replace'
441
(t, stack) = self._setup_test_stack(stack_name)
443
server = servers.Server('server_server_update_flavor_replace',
444
t['Resources']['WebServer'], stack)
446
update_template = copy.deepcopy(server.t)
447
# confirm that when flavor_update_policy is changed during
448
# the update then the updated policy is followed for a flavor
450
update_template['Properties']['flavor_update_policy'] = 'REPLACE'
451
update_template['Properties']['flavor'] = 'm1.smigish'
452
updater = scheduler.TaskRunner(server.update, update_template)
453
self.assertRaises(resource.UpdateReplace, updater)
455
def test_server_update_replace(self):
456
return_server = self.fc.servers.list()[1]
457
server = self._create_test_server(return_server,
458
'test_server_update')
460
update_template = copy.deepcopy(server.t)
461
update_template['Notallowed'] = {'test': 123}
462
updater = scheduler.TaskRunner(server.update, update_template)
463
self.assertRaises(resource.UpdateReplace, updater)
465
def test_server_update_properties(self):
466
return_server = self.fc.servers.list()[1]
467
server = self._create_test_server(return_server,
468
'test_server_update')
470
update_template = copy.deepcopy(server.t)
471
update_template['Properties']['key_name'] = 'mustreplace'
472
updater = scheduler.TaskRunner(server.update, update_template)
473
self.assertRaises(resource.UpdateReplace, updater)
475
def test_server_status_build(self):
476
return_server = self.fc.servers.list()[0]
477
server = self._setup_test_server(return_server,
478
'test_server_status_build')
479
server.resource_id = 1234
481
# Bind fake get method which Server.check_create_complete will call
482
def activate_status(server):
483
server.status = 'ACTIVE'
484
return_server.get = activate_status.__get__(return_server)
487
scheduler.TaskRunner(server.create)()
488
self.assertEqual(server.state, (server.CREATE, server.COMPLETE))
490
def test_server_status_suspend_no_resource_id(self):
491
return_server = self.fc.servers.list()[1]
492
server = self._create_test_server(return_server,
493
'test_server_suspend')
495
server.resource_id = None
498
ex = self.assertRaises(exception.ResourceFailure,
499
scheduler.TaskRunner(server.suspend))
500
self.assertEqual('Error: Cannot suspend test_server_suspend_name, '
501
'resource_id not set',
503
self.assertEqual(server.state, (server.SUSPEND, server.FAILED))
507
def test_server_status_suspend_not_found(self):
508
return_server = self.fc.servers.list()[1]
509
server = self._create_test_server(return_server,
510
'test_server_suspend')
512
server.resource_id = 1234
513
self.m.StubOutWithMock(self.fc.client, 'get_servers_1234')
514
get = self.fc.client.get_servers_1234
515
get().AndRaise(servers.clients.novaclient.exceptions.NotFound(404))
519
ex = self.assertRaises(exception.ResourceFailure,
520
scheduler.TaskRunner(server.suspend))
521
self.assertEqual('NotFound: Failed to find server 1234',
523
self.assertEqual(server.state, (server.SUSPEND, server.FAILED))
527
def test_server_status_suspend_immediate(self):
528
return_server = self.fc.servers.list()[1]
529
server = self._create_test_server(return_server,
530
'test_server_suspend')
532
server.resource_id = 1234
535
# Override the get_servers_1234 handler status to SUSPENDED
536
d = {'server': self.fc.client.get_servers_detail()[1]['servers'][0]}
537
d['server']['status'] = 'SUSPENDED'
538
self.m.StubOutWithMock(self.fc.client, 'get_servers_1234')
539
get = self.fc.client.get_servers_1234
540
get().AndReturn((200, d))
543
scheduler.TaskRunner(server.suspend)()
544
self.assertEqual(server.state, (server.SUSPEND, server.COMPLETE))
548
def test_server_status_resume_immediate(self):
549
return_server = self.fc.servers.list()[1]
550
server = self._create_test_server(return_server,
551
'test_server_resume')
553
server.resource_id = 1234
556
# Override the get_servers_1234 handler status to SUSPENDED
557
d = {'server': self.fc.client.get_servers_detail()[1]['servers'][0]}
558
d['server']['status'] = 'ACTIVE'
559
self.m.StubOutWithMock(self.fc.client, 'get_servers_1234')
560
get = self.fc.client.get_servers_1234
561
get().AndReturn((200, d))
563
server.state_set(server.SUSPEND, server.COMPLETE)
565
scheduler.TaskRunner(server.resume)()
566
self.assertEqual(server.state, (server.RESUME, server.COMPLETE))
570
def test_server_status_suspend_wait(self):
571
return_server = self.fc.servers.list()[1]
572
server = self._create_test_server(return_server,
573
'test_server_suspend')
575
server.resource_id = 1234
578
# Override the get_servers_1234 handler status to SUSPENDED, but
579
# return the ACTIVE state first (twice, so we sleep)
580
d1 = {'server': self.fc.client.get_servers_detail()[1]['servers'][0]}
581
d2 = copy.deepcopy(d1)
582
d1['server']['status'] = 'ACTIVE'
583
d2['server']['status'] = 'SUSPENDED'
584
self.m.StubOutWithMock(self.fc.client, 'get_servers_1234')
585
get = self.fc.client.get_servers_1234
586
get().AndReturn((200, d1))
587
get().AndReturn((200, d1))
588
get().AndReturn((200, d2))
591
scheduler.TaskRunner(server.suspend)()
592
self.assertEqual(server.state, (server.SUSPEND, server.COMPLETE))
596
def test_server_status_suspend_unknown_status(self):
597
return_server = self.fc.servers.list()[1]
598
server = self._create_test_server(return_server,
599
'test_server_suspend')
601
server.resource_id = 1234
604
# Override the get_servers_1234 handler status to SUSPENDED, but
605
# return the ACTIVE state first (twice, so we sleep)
606
d1 = {'server': self.fc.client.get_servers_detail()[1]['servers'][0]}
607
d2 = copy.deepcopy(d1)
608
d1['server']['status'] = 'ACTIVE'
609
d2['server']['status'] = 'TRANSMOGRIFIED'
610
self.m.StubOutWithMock(self.fc.client, 'get_servers_1234')
611
get = self.fc.client.get_servers_1234
612
get().AndReturn((200, d1))
613
get().AndReturn((200, d1))
614
get().AndReturn((200, d2))
617
ex = self.assertRaises(exception.ResourceFailure,
618
scheduler.TaskRunner(server.suspend))
619
self.assertEqual('Error: Suspend of server sample-server failed '
620
'with unknown status: TRANSMOGRIFIED',
622
self.assertEqual(server.state, (server.SUSPEND, server.FAILED))
626
def test_server_status_resume_wait(self):
627
return_server = self.fc.servers.list()[1]
628
server = self._create_test_server(return_server,
629
'test_server_resume')
631
server.resource_id = 1234
634
# Override the get_servers_1234 handler status to ACTIVE, but
635
# return the SUSPENDED state first (twice, so we sleep)
636
d1 = {'server': self.fc.client.get_servers_detail()[1]['servers'][0]}
637
d2 = copy.deepcopy(d1)
638
d1['server']['status'] = 'SUSPENDED'
639
d2['server']['status'] = 'ACTIVE'
640
self.m.StubOutWithMock(self.fc.client, 'get_servers_1234')
641
get = self.fc.client.get_servers_1234
642
get().AndReturn((200, d1))
643
get().AndReturn((200, d1))
644
get().AndReturn((200, d2))
647
server.state_set(server.SUSPEND, server.COMPLETE)
649
scheduler.TaskRunner(server.resume)()
650
self.assertEqual(server.state, (server.RESUME, server.COMPLETE))
654
def test_server_status_resume_no_resource_id(self):
655
return_server = self.fc.servers.list()[1]
656
server = self._create_test_server(return_server,
657
'test_server_suspend')
659
server.resource_id = None
662
server.state_set(server.SUSPEND, server.COMPLETE)
663
ex = self.assertRaises(exception.ResourceFailure,
664
scheduler.TaskRunner(server.resume))
665
self.assertEqual('Error: Cannot resume test_server_suspend_name, '
666
'resource_id not set',
668
self.assertEqual(server.state, (server.RESUME, server.FAILED))
672
def test_server_status_resume_not_found(self):
673
return_server = self.fc.servers.list()[1]
674
server = self._create_test_server(return_server,
675
'test_server_resume')
677
server.resource_id = 1234
680
# Override the get_servers_1234 handler status to ACTIVE, but
681
# return the SUSPENDED state first (twice, so we sleep)
682
self.m.StubOutWithMock(self.fc.client, 'get_servers_1234')
683
get = self.fc.client.get_servers_1234
684
get().AndRaise(servers.clients.novaclient.exceptions.NotFound(404))
687
server.state_set(server.SUSPEND, server.COMPLETE)
689
ex = self.assertRaises(exception.ResourceFailure,
690
scheduler.TaskRunner(server.resume))
691
self.assertEqual('NotFound: Failed to find server 1234',
693
self.assertEqual(server.state, (server.RESUME, server.FAILED))
697
def test_server_status_build_spawning(self):
698
self._test_server_status_not_build_active('BUILD(SPAWNING)')
700
def test_server_status_hard_reboot(self):
701
self._test_server_status_not_build_active('HARD_REBOOT')
703
def test_server_status_password(self):
704
self._test_server_status_not_build_active('PASSWORD')
706
def test_server_status_reboot(self):
707
self._test_server_status_not_build_active('REBOOT')
709
def test_server_status_rescue(self):
710
self._test_server_status_not_build_active('RESCUE')
712
def test_server_status_resize(self):
713
self._test_server_status_not_build_active('RESIZE')
715
def test_server_status_revert_resize(self):
716
self._test_server_status_not_build_active('REVERT_RESIZE')
718
def test_server_status_shutoff(self):
719
self._test_server_status_not_build_active('SHUTOFF')
721
def test_server_status_suspended(self):
722
self._test_server_status_not_build_active('SUSPENDED')
724
def test_server_status_verify_resize(self):
725
self._test_server_status_not_build_active('VERIFY_RESIZE')
727
def _test_server_status_not_build_active(self, uncommon_status):
728
return_server = self.fc.servers.list()[0]
729
server = self._setup_test_server(return_server,
730
'test_server_status_build')
731
server.resource_id = 1234
733
check_iterations = [0]
735
# Bind fake get method which Server.check_create_complete will call
736
def activate_status(server):
737
check_iterations[0] += 1
738
if check_iterations[0] == 1:
739
server.status = uncommon_status
740
if check_iterations[0] > 2:
741
server.status = 'ACTIVE'
742
return_server.get = activate_status.__get__(return_server)
745
scheduler.TaskRunner(server.create)()
746
self.assertEqual(server.state, (server.CREATE, server.COMPLETE))
750
def test_build_nics(self):
751
self.assertEqual(None, servers.Server._build_nics([]))
752
self.assertEqual(None, servers.Server._build_nics(None))
754
{'net-id': '1234abcd'},
755
{'v4-fixed-ip': '192.0.2.0'},
756
{'port-id': 'aaaabbbb'}
757
], servers.Server._build_nics([
758
{'uuid': '1234abcd'},
759
{'fixed_ip': '192.0.2.0'},
763
def test_server_without_ip_address(self):
764
return_server = self.fc.servers.list()[3]
765
server = self._create_test_server(return_server,
766
'test_without_ip_address')
768
self.assertEqual(server.FnGetAtt('addresses'), {'empty_net': []})
769
self.assertEqual(server.FnGetAtt('networks'), {'empty_net': []})
770
self.assertEqual(server.FnGetAtt('first_private_address'), '')
771
self.assertEqual(server.FnGetAtt('first_public_address'), '')
773
def test_build_block_device_mapping(self):
775
None, servers.Server._build_block_device_mapping([]))
777
None, servers.Server._build_block_device_mapping(None))
781
'vdb': ['1234', 'snap'],
782
}, servers.Server._build_block_device_mapping([
783
{'device_name': 'vda', 'volume_id': '1234'},
784
{'device_name': 'vdb', 'snapshot_id': '1234'},
788
'vdc': ['1234', '', 10],
789
'vdd': ['1234', 'snap', 0, True]
790
}, servers.Server._build_block_device_mapping([
792
'device_name': 'vdc',
797
'device_name': 'vdd',
798
'snapshot_id': '1234',
799
'delete_on_termination': True
added
removed
heat/tests/test_server.py
