blob: 5e27194adf3f1482937e117d2e6a0be9da7351fb [file] [log] [blame]
import base64
import json
import os
from kong import openstack
from kong import exceptions
from kong import tests
from kong.common import ssh
class ServersTest(tests.FunctionalTest):
def setUp(self):
super(ServersTest, self).setUp()
self.os = openstack.Manager(self.nova)
self.image_ref = self.glance['image_id']
self.flavor_ref = self.nova['flavor_ref']
self.ssh_timeout = self.nova['ssh_timeout']
self.build_timeout = self.nova['build_timeout']
def tearDown(self):
if getattr(self, 'server_id', False):
self.os.nova.delete_server(self.server_id)
def _assert_server_entity(self, server):
actual_keys = set(server.keys())
expected_keys = set((
'id',
'name',
'hostId',
'status',
'metadata',
'addresses',
'links',
'progress',
'image',
'flavor',
'created',
'updated',
'accessIPv4',
'accessIPv6',
#KNOWN-ISSUE lp804093
'uuid',
))
self.assertTrue(expected_keys <= actual_keys)
server_id = str(server['id'])
host = self.nova['host']
port = self.nova['port']
api_url = '%s:%s' % (host, port)
base_url = os.path.join(api_url, self.nova['apiver'])
self_link = 'http://' + os.path.join(base_url,
self.os.nova.project_id,
'servers', server_id)
bookmark_link = 'http://' + os.path.join(api_url,
self.os.nova.project_id,
'servers', server_id)
expected_links = [
{
'rel': 'self',
'href': self_link,
},
{
'rel': 'bookmark',
'href': bookmark_link,
},
]
self.assertEqual(server['links'], expected_links)
def test_build_server(self):
"""Build a server"""
expected_server = {
'name': 'testserver',
'metadata': {
'key1': 'value1',
'key2': 'value2',
},
'imageRef': self.image_ref,
'flavorRef': self.flavor_ref,
}
post_body = json.dumps({'server': expected_server})
response, body = self.os.nova.request('POST',
'/servers',
body=post_body)
self.assertEqual(response.status, 202)
_body = json.loads(body)
self.assertEqual(_body.keys(), ['server'])
created_server = _body['server']
self.server_id = created_server['id'] # for the tearDown
admin_pass = created_server.pop('adminPass')
self._assert_server_entity(created_server)
self.assertEqual(expected_server['name'], created_server['name'])
self.assertEqual(expected_server['metadata'],
created_server['metadata'])
self.os.nova.wait_for_server_status(created_server['id'],
'ACTIVE',
timeout=self.build_timeout)
server = self.os.nova.get_server(created_server['id'])
# Find IP of server
try:
(_, network) = server['addresses'].popitem()
ip = network[0]['addr']
except KeyError:
self.fail("Failed to retrieve IP address from server entity")
# Assert password works
if int(self.nova['ssh_timeout']) > 0:
client = ssh.Client(ip, 'root', admin_pass, self.ssh_timeout)
self.assertTrue(client.test_connection_auth())
test_build_server.tags = ['nova', 'glance']
def test_build_server_with_file(self):
"""Build a server with an injected file"""
file_contents = 'testing'
expected_server = {
'name': 'testserver',
'metadata': {
'key1': 'value1',
'key2': 'value2',
},
'personality': [
{
'path': '/etc/test.txt',
'contents': base64.b64encode(file_contents),
},
],
'imageRef': self.image_ref,
'flavorRef': self.flavor_ref,
}
post_body = json.dumps({'server': expected_server})
response, body = self.os.nova.request('POST',
'/servers',
body=post_body)
self.assertEqual(response.status, 202)
_body = json.loads(body)
self.assertEqual(_body.keys(), ['server'])
created_server = _body['server']
self.server_id = _body['server']['id']
admin_pass = created_server.pop('adminPass', None)
self._assert_server_entity(created_server)
self.assertEqual(expected_server['name'], created_server['name'])
self.assertEqual(expected_server['metadata'],
created_server['metadata'])
self.os.nova.wait_for_server_status(created_server['id'],
'ACTIVE',
timeout=self.build_timeout)
server = self.os.nova.get_server(created_server['id'])
# Find IP of server
try:
(_, network) = server['addresses'].popitem()
ip = network[0]['addr']
except KeyError:
self.fail("Failed to retrieve IP address from server entity")
# Assert injected file is on instance, also verifying password works
if int(self.nova['ssh_timeout']) > 0:
client = ssh.Client(ip, 'root', admin_pass, self.ssh_timeout)
injected_file = client.exec_command('cat /etc/test.txt')
self.assertEqual(injected_file, file_contents)
test_build_server_with_file.tags = ['nova', 'glance']
def test_build_server_with_password(self):
"""Build a server with a password"""
server_password = 'testpwd'
expected_server = {
'name': 'testserver',
'metadata': {
'key1': 'value1',
'key2': 'value2',
},
'adminPass': server_password,
'imageRef': self.image_ref,
'flavorRef': self.flavor_ref,
}
post_body = json.dumps({'server': expected_server})
response, body = self.os.nova.request('POST',
'/servers',
body=post_body)
self.assertEqual(response.status, 202)
_body = json.loads(body)
self.assertEqual(_body.keys(), ['server'])
created_server = _body['server']
admin_pass = created_server.pop('adminPass', None)
self._assert_server_entity(created_server)
self.assertEqual(expected_server['name'], created_server['name'])
self.assertEqual(expected_server['adminPass'], admin_pass)
self.assertEqual(expected_server['metadata'],
created_server['metadata'])
self.os.nova.wait_for_server_status(created_server['id'],
'ACTIVE',
timeout=self.build_timeout)
server = self.os.nova.get_server(created_server['id'])
# Find IP of server
try:
(_, network) = server['addresses'].popitem()
ip = network[0]['addr']
except KeyError:
self.fail("Failed to retrieve IP address from server entity")
# Assert password was set to that in request ( if ssh_timeout is > 0
if int(self.nova['ssh_timeout']) > 0:
client = ssh.Client(ip, 'root', server_password, self.ssh_timeout)
self.assertTrue(client.test_connection_auth())
test_build_server_with_password.tags = ['nova', 'glance']
def test_delete_server_building(self):
"""Delete a server while building"""
# Make create server request
server = {
'name' : 'testserver',
'imageRef' : self.image_ref,
'flavorRef' : self.flavor_ref,
}
created_server = self.os.nova.create_server(server)
# Server should immediately be accessible, but in have building status
server = self.os.nova.get_server(created_server['id'])
self.assertEqual(server['status'], 'BUILD')
self.os.nova.delete_server(created_server['id'])
# Poll server until deleted
try:
url = '/servers/%s' % created_server['id']
self.os.nova.poll_request_status('GET', url, 404)
except exceptions.TimeoutException:
self.fail("Server deletion timed out")
test_delete_server_building.tags = ['nova', 'glance']
def test_delete_server_active(self):
"""Delete a server after fully built"""
expected_server = {
'name' : 'testserver',
'imageRef' : self.image_ref,
'flavorRef' : self.flavor_ref,
}
created_server = self.os.nova.create_server(expected_server)
server_id = created_server['id']
self.os.nova.wait_for_server_status(server_id,
'ACTIVE',
timeout=self.build_timeout)
self.os.nova.delete_server(server_id)
# Poll server until deleted
try:
url = '/servers/%s' % server_id
self.os.nova.poll_request_status('GET', url, 404)
except exceptions.TimeoutException:
self.fail("Server deletion timed out")
test_delete_server_active.tags = ['nova', 'glance']
def test_update_server_name(self):
"""Change the name of a server"""
expected_server = {
'name' : 'testserver',
'imageRef' : self.image_ref,
'flavorRef' : self.flavor_ref,
}
created_server = self.os.nova.create_server(expected_server)
self.assertTrue(expected_server['name'], created_server['name'])
server_id = created_server['id']
# Wait for it to be built
self.os.nova.wait_for_server_status(server_id,
'ACTIVE',
timeout=self.build_timeout)
# Update name
new_server = {'name': 'updatedtestserver'}
put_body = json.dumps({
'server': new_server,
})
url = '/servers/%s' % server_id
resp, body = self.os.nova.request('PUT', url, body=put_body)
self.assertEqual(resp.status, 200)
data = json.loads(body)
self.assertEqual(data.keys(), ['server'])
self._assert_server_entity(data['server'])
self.assertEqual('updatedtestserver', data['server']['name'])
# Get Server information
resp, body = self.os.nova.request('GET', '/servers/%s' % server_id)
self.assertEqual(200, resp.status)
data = json.loads(body)
self.assertEqual(data.keys(), ['server'])
self._assert_server_entity(data['server'])
self.assertEqual('updatedtestserver', data['server']['name'])
self.os.nova.delete_server(server_id)
test_update_server_name.tags = ['nova', 'glance']
def test_create_server_invalid_image(self):
"""Create a server with an unknown image"""
post_body = json.dumps({
'server' : {
'name' : 'testserver',
'imageRef' : -1,
'flavorRef' : self.flavor_ref,
}
})
resp, body = self.os.nova.request('POST', '/servers', body=post_body)
self.assertEqual(400, resp.status)
fault = json.loads(body)
expected_fault = {
"badRequest": {
"message": "Cannot find requested image",
"code": 400,
},
}
# KNOWN-ISSUE - The error message is confusing and should be improved
#self.assertEqual(fault, expected_fault)
test_create_server_invalid_image.tags = ['nova', 'glance']
def test_create_server_invalid_flavor(self):
"""Create a server with an unknown flavor"""
post_body = json.dumps({
'server' : {
'name' : 'testserver',
'imageRef' : self.image_ref,
'flavorRef' : -1,
}
})
resp, body = self.os.nova.request('POST', '/servers', body=post_body)
self.assertEqual(400, resp.status)
fault = json.loads(body)
expected_fault = {
"badRequest": {
"message": "Cannot find requested flavor",
"code": 400,
},
}
# KNOWN-ISSUE lp804084
#self.assertEqual(fault, expected_fault)
test_create_server_invalid_flavor.tags = ['nova', 'glance']