1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright (c) 2010 OpenStack, LLC.
4
# Administrator of the National Aeronautics and Space Administration.
7
# Licensed under the Apache License, Version 2.0 (the "License"); you may
8
# not use this file except in compliance with the License. You may obtain
9
# a copy of the License at
11
# http://www.apache.org/licenses/LICENSE-2.0
13
# Unless required by applicable law or agreed to in writing, software
14
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
15
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
16
# License for the specific language governing permissions and limitations
19
"""Tests For Console proxy."""
21
from nova import context
23
from nova import exception
24
from nova import flags
25
from nova.openstack.common import importutils
29
flags.DECLARE('console_driver', 'nova.console.manager')
32
class ConsoleTestCase(test.TestCase):
33
"""Test case for console proxy"""
35
super(ConsoleTestCase, self).setUp()
36
self.flags(console_driver='nova.console.fake.FakeConsoleProxy',
38
self.console = importutils.import_object(FLAGS.console_manager)
40
self.project_id = 'fake'
41
self.context = context.RequestContext(self.user_id, self.project_id)
42
self.host = 'test_compute_host'
44
def _create_instance(self):
45
"""Create a test instance"""
47
#inst['host'] = self.host
48
#inst['name'] = 'instance-1234'
50
inst['reservation_id'] = 'r-fakeres'
51
inst['launch_time'] = '10'
52
inst['user_id'] = self.user_id
53
inst['project_id'] = self.project_id
54
inst['instance_type_id'] = 1
55
inst['ami_launch_index'] = 0
56
return db.instance_create(self.context, inst)['id']
58
def test_get_pool_for_instance_host(self):
59
pool = self.console.get_pool_for_instance_host(self.context,
61
self.assertEqual(pool['compute_host'], self.host)
63
def test_get_pool_creates_new_pool_if_needed(self):
64
self.assertRaises(exception.NotFound,
65
db.console_pool_get_by_host_type,
69
self.console.driver.console_type)
70
pool = self.console.get_pool_for_instance_host(self.context,
72
pool2 = db.console_pool_get_by_host_type(self.context,
75
self.console.driver.console_type)
76
self.assertEqual(pool['id'], pool2['id'])
78
def test_get_pool_does_not_create_new_pool_if_exists(self):
79
pool_info = {'address': '127.0.0.1',
81
'password': '1234pass',
82
'host': self.console.host,
83
'console_type': self.console.driver.console_type,
84
'compute_host': 'sometesthostname'}
85
new_pool = db.console_pool_create(self.context, pool_info)
86
pool = self.console.get_pool_for_instance_host(self.context,
88
self.assertEqual(pool['id'], new_pool['id'])
90
def test_add_console(self):
91
instance_id = self._create_instance()
92
self.console.add_console(self.context, instance_id)
93
instance = db.instance_get(self.context, instance_id)
94
pool = db.console_pool_get_by_host_type(self.context,
95
instance['host'], self.console.host,
96
self.console.driver.console_type)
98
console_instances = [con['instance_id'] for con in pool.consoles]
99
self.assert_(instance_id in console_instances)
100
db.instance_destroy(self.context, instance_id)
102
def test_add_console_does_not_duplicate(self):
103
instance_id = self._create_instance()
104
cons1 = self.console.add_console(self.context, instance_id)
105
cons2 = self.console.add_console(self.context, instance_id)
106
self.assertEqual(cons1, cons2)
107
db.instance_destroy(self.context, instance_id)
109
def test_remove_console(self):
110
instance_id = self._create_instance()
111
console_id = self.console.add_console(self.context, instance_id)
112
self.console.remove_console(self.context, console_id)
114
self.assertRaises(exception.NotFound,
118
db.instance_destroy(self.context, instance_id)