openstack
/
nova
Code Issues Proposed changes

vnc unit-test fixes 

Change-Id: I6097e45fb2c52a53ac37c6b51abdb432c722e47b
This commit is contained in:
Loganathan Parthipan 2013-02-26 22:20:50 +00:00
parent 65e36a1c28
commit e02b45976f
4 changed files with 31 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
nova/consoleauth/manager.py
View File

@ -53,6 +53,7 @@ class ConsoleAuthManager(manager.Manager):
super(ConsoleAuthManager, self).__init__(*args, **kwargs)
self.compute_api = compute.API()
self.tokens = {}
self.instance_tokens = {}
utils.LoopingCall(self._delete_expired_tokens).start(1)
def _delete_expired_tokens(self):
@ -67,28 +68,28 @@ class ConsoleAuthManager(manager.Manager):
del self.tokens[k]
def authorize_console(self, context, token, console_type, host, port,
instance_id, internal_access_path):
internal_access_path, instance_id=None):
self.tokens[token] = {'token': token,
'instance_id': instance_id,
'console_type': console_type,
'host': host,
'port': port,
'instance_id': instance_id,
'internal_access_path': internal_access_path,
'last_activity_at': time.time()}
token_dict = self.tokens[token]
if instance_id is not None:
tokens = self.tokens[instance_id]
tokens.append(token)
self.tokens[instance_id] = tokens
instance_token_list = self.instance_tokens.get(instance_id, [])
instance_token_list.append(token)
self.instance_tokens[instance_id] = instance_token_list
LOG.audit(_("Received Token: %(token)s, %(token_dict)s)"), locals())
def _validate_console(self, context, token):
def _validate_console(self, token):
console_valid = False
token_dict = self.tokens[token]
try:
console_valid = self.compute_api.validate_vnc_console(context,
token_dict['instance_id'],
token_dict['instance_uuid'],
token_dict['host'],
token_dict['port'])
except exception.InstanceNotFound:
@ -98,12 +99,12 @@ class ConsoleAuthManager(manager.Manager):
def check_token(self, context, token):
token_valid = token in self.tokens
LOG.audit(_("Checking Token: %(token)s, %(token_valid)s)"), locals())
if token_valid and _validate_console(token):
if token_valid and self._validate_console(token):
return self.tokens[token]
def delete_tokens_for_instance(self, context, instance_id):
for token in self.tokens[instance_id]:
for token in self.instance_tokens[instance_id]:
token_dict = self.tokens[token]
token_dict['last_activity_at'] = 0
self.tokens[token] = token_dict
del self.tokens[instance_id]
del self.instance_tokens[instance_id]

1
nova/tests/policy.json
View File

@ -22,6 +22,7 @@
"compute:get_vnc_console": [],
"compute:get_console_output": [],
"compute:validate_vnc_console": [],
"compute:associate_floating_ip": [],
"compute:reset_network": [],

25
nova/tests/test_compute.py
View File

@ -197,6 +197,7 @@ class ComputeTestCase(BaseTestCase):
fake_get_nw_info)
self.stubs.Set(nova.network.API, 'allocate_for_instance',
fake_get_nw_info)
self.compute_api = compute.API()
def test_wrap_instance_fault(self):
inst_uuid = "fake_uuid"
@ -760,15 +761,15 @@ class ComputeTestCase(BaseTestCase):
vnc = {'host': 'myhost', 'port': '5900'}
return vnc
self.stubs.Set(self.compute_api, '_call_compute_message', fake)
instance = self._create_instance_full()
self.compute.run_instance(self.context, instance['id'])
instance = self._create_fake_instance()
self.compute.run_instance(self.context, instance['uuid'])
console_valid = self.compute_api.validate_vnc_console(self.context,
instance['id'],
instance['uuid'],
'myhost',
'5900')
self.assertTrue(console_valid)
self.compute.terminate_instance(self.context, instance['id'])
self.compute.terminate_instance(self.context, instance['uuid'])
def test_validate_vnc_console_wrong_port(self):
"""Check if a vnc console is really for the instance"""
@ -776,24 +777,24 @@ class ComputeTestCase(BaseTestCase):
vnc = {'host': 'myhost', 'port': '5901'}
return vnc
self.stubs.Set(self.compute_api, '_call_compute_message', fake)
instance = self._create_instance_full()
self.compute.run_instance(self.context, instance['id'])
instance = self._create_fake_instance()
self.compute.run_instance(self.context, instance['uuid'])
console_valid = self.compute_api.validate_vnc_console(self.context,
instance['id'],
instance['uuid'],
'myhost',
'5900')
self.assertFalse(console_valid)
self.compute.terminate_instance(self.context, instance['id'])
self.compute.terminate_instance(self.context, instance['uuid'])
def test_validate_vnc_console_deleted_instance(self):
"""Check if a vnc console is really for the instance"""
instance = self._create_instance_full()
self.compute.run_instance(self.context, instance['id'])
instance = self._create_fake_instance()
self.compute.run_instance(self.context, instance['uuid'])
self.assertRaises(exception.InstanceNotFound,
self.compute_api.validate_vnc_console,
self.context, 5555, 'myhost', '5900')
self.compute.terminate_instance(self.context, instance['id'])
self.compute.terminate_instance(self.context, instance['uuid'])
def test_xvpvnc_vnc_console(self):
"""Make sure we can a vnc console for an instance."""
@ -3125,10 +3126,12 @@ class ComputeAPITestCase(BaseTestCase):
"""Make sure we can a vnc console for an instance."""
fake_instance = {'uuid': 'fake_uuid',
'id': 'fake_id',
'host': 'fake_compute_host'}
fake_console_type = "novnc"
fake_connect_info = {'token': 'fake_token',
'console_type': fake_console_type,
'instance_id': fake_instance['id'],
'host': 'fake_console_host',
'port': 'fake_console_port',
'internal_access_path': 'fake_access_path',

7
nova/tests/test_consoleauth.py
View File

@ -46,6 +46,7 @@ class ConsoleauthTestCase(test.TestCase):
def test_tokens_expire(self):
"""Test that tokens expire correctly."""
token = 'mytok'
self.flags(console_token_ttl=1)
def fake_validate_console(*args, **kwargs):
@ -54,8 +55,10 @@ class ConsoleauthTestCase(test.TestCase):
"_validate_console",
fake_validate_console)
self.manager.authorize_console(self.context, token, 'novnc',
'127.0.0.1', 'host', '')
self.manager.authorize_console(self.context, token,
'novnc',
'127.0.0.1', '5900', '', 123)
self.assertTrue(self.manager.check_token(self.context, token))
time.sleep(1.1)
self.manager._delete_expired_tokens()
self.assertFalse(self.manager.check_token(self.context, token))