Commit 1d8318cd authored by Ilias Tsitsimpis's avatar Ilias Tsitsimpis
Browse files

burnin: Finish ServerTestSuite

parent 8a0c1c58
......@@ -39,8 +39,11 @@ had grown too much.
"""
import time
import base64
import socket
import random
import paramiko
import tempfile
import subprocess
from synnefo_tools.burnin.common import BurninTests
......@@ -87,13 +90,13 @@ class CycladesTests(BurninTests):
server['name'], server['id'])
return self.clients.cyclades.get_server_details(server['id'])
def _create_server(self, name, image, flavor):
def _create_server(self, name, image, flavor, personality):
"""Create a new server"""
self.info("Creating a server with name %s", name)
self.info("Using image %s with id %s", image['name'], image['id'])
self.info("Using flavor %s with id %s", flavor['name'], flavor['id'])
server = self.clients.cyclades.create_server(
name, flavor['id'], image['id'])
name, flavor['id'], image['id'], personality=personality)
self.info("Server id: %s", server['id'])
self.info("Server password: %s", server['adminPass'])
......@@ -178,7 +181,7 @@ class CycladesTests(BurninTests):
opmsg = opmsg % (familystr.get(family, "Unknown"), host, port)
return self._try_until_timeout_expires(opmsg, check_fun)
def _get_ip(self, server, version):
def _get_ip(self, server, version=4):
"""Get the public IP of a server from the detailed server info"""
assert version in (4, 6)
......@@ -190,15 +193,14 @@ class CycladesTests(BurninTests):
public_addrs = nic['ipv' + str(version)]
self.assertIsNotNone(public_addrs)
msg = "Servers %s public IPv%s is %s"
self.info(msg, server['name'], version, public_addrs)
msg = "Server's public IPv%s is %s"
self.info(msg, version, public_addrs)
return public_addrs
def _insist_on_ping(self, ip_addr, version):
def _insist_on_ping(self, ip_addr, version=4):
"""Test server responds to a single IPv4 of IPv6 ping"""
def check_fun():
"""Ping to server"""
assert version in (4, 6)
cmd = ("ping%s -c 3 -w 20 %s" %
("6" if version == 6 else "", ip_addr))
ping = subprocess.Popen(
......@@ -208,11 +210,69 @@ class CycladesTests(BurninTests):
ret = ping.wait()
if ret != 0:
raise Retry
assert version in (4, 6)
opmsg = "Sent IPv%s ping requests to %s"
self.info(opmsg, version, ip_addr)
opmsg = opmsg % (version, ip_addr)
self._try_until_timeout_expires(opmsg, check_fun)
def _image_is(self, image, osfamily):
"""Return true if the image is of `osfamily'"""
d_image = self.clients.cyclades.get_image_details(image['id'])
return d_image['metadata']['osfamily'].lower().find(osfamily) >= 0
def _ssh_execute(self, hostip, username, password, command):
"""Execute a command via ssh"""
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(hostip, username=username, password=password)
except socket.error as err:
self.fail(err)
try:
_, stdout, _ = ssh.exec_command(command)
except paramiko.SSHException as err:
self.fail(err)
status = stdout.channel.recv_exit_status()
output = stdout.readlines()
ssh.close()
return output, status
def _insist_get_hostname_over_ssh(self, hostip, username, password):
"""Connect to server using ssh and get it's hostname"""
def check_fun():
"""Get hostname"""
try:
lines, status = self._ssh_execute(
hostip, username, password, "hostname")
self.assertEqual(status, 0)
self.assertEqual(len(lines), 1)
# Remove new line
return lines[0].strip('\n')
except AssertionError:
raise Retry()
opmsg = "Connecting to server using ssh and get it's hostname"
self.info(opmsg)
hostname = self._try_until_timeout_expires(opmsg, check_fun)
self.info("Server's hostname is %s", hostname)
return hostname
# Too many arguments. pylint: disable-msg=R0913
def _check_file_through_ssh(self, hostip, username, password,
remotepath, content):
"""Fetch file from server and compare contents"""
self.info("Fetching file %s from remote server", remotepath)
transport = paramiko.Transport((hostip, 22))
transport.connect(username=username, password=password)
with tempfile.NamedTemporaryFile() as ftmp:
sftp = paramiko.SFTPClient.from_transport(transport)
sftp.get(remotepath, ftmp.name)
sftp.close()
transport.close()
self.info("Comparing file contents")
remote_content = base64.b64encode(ftmp.read())
self.assertEqual(content, remote_content)
class Retry(Exception):
"""Retry the action
......
......@@ -158,7 +158,7 @@ class ImagesTestSuite(BurninTests):
os.path.join(self.temp_dir, self.temp_image_name)
# Write to file
self.info("Download image to %s", self.temp_image_file)
self.info("Downloading image to %s", self.temp_image_file)
with open(self.temp_image_file, "w+b") as fout:
self.clients.pithos.download_object(image_name, fout)
......
......@@ -38,6 +38,8 @@ This is the burnin class that tests the Servers' functionality
import sys
import IPy
import stat
import base64
import random
import socket
......@@ -52,19 +54,34 @@ from synnefo_tools.burnin.cyclades_common import CycladesTests
class GeneratedServerTestSuite(CycladesTests):
"""Test Spawning Serverfunctionality"""
use_image = Proper(value=None)
personality = Proper(value=None)
avail_flavors = Proper(value=None)
use_flavor = Proper(value=None)
server = Proper(value=None)
ipv4 = Proper(value=None)
ipv6 = Proper(value=None)
username = Proper(value=None)
password = Proper(value=None)
def test_001_submit_create_server(self):
"""Submit a create server request"""
if self._image_is(self.use_image, "linux"):
# Enforce personality test
self.info("Creating personality content to be used")
self.personality = [{
'path': "/root/test_inj_file",
'owner': "root",
'group': "root",
'mode': stat.S_IRUSR | stat.S_IWUSR,
'contents': base64.b64encode("This is a personality file")
}]
servername = "%s for %s" % (self.run_id, self.use_image['name'])
self.use_flavor = random.choice(self.avail_flavors)
self.server = self._create_server(
servername, self.use_image, self.use_flavor)
servername, self.use_image, self.use_flavor, self.personality)
self.username = self._get_connection_username(self.server)
self.password = self.server['adminPass']
def test_002_server_build_list(self):
"""Test server is in BUILD state, in server list"""
......@@ -147,24 +164,24 @@ class GeneratedServerTestSuite(CycladesTests):
# Update the server attribute
self.server = server
self.ipv4 = self._get_ip(server, 4)
self.ipv4 = self._get_ip(server, version=4)
self.assertEquals(IPy.IP(self.ipv4).version(), 4)
def test_008_server_has_ipv6(self):
"""Test active server has a valid IPv6 address"""
self._skip_if(not self.use_ipv6, "--no-ipv6 flag enabled")
self.ipv6 = self._get_ip(self.server, 6)
self.ipv6 = self._get_ip(self.server, version=6)
self.assertEquals(IPy.IP(self.ipv6).version(), 6)
def test_009_server_ping_ipv4(self):
"""Test server responds to ping on IPv4 address"""
self._insist_on_ping(self.ipv4, 4)
self._insist_on_ping(self.ipv4, version=4)
def test_010_server_ping_ipv6(self):
"""Test server responds to ping on IPv6 address"""
self._skip_if(not self.use_ipv6, "--no-ipv6 flag enabled")
self._insist_on_ping(self.ipv6, 6)
self._insist_on_ping(self.ipv6, version=6)
def test_011_submit_shutdown(self):
"""Test submit request to shutdown server"""
......@@ -186,6 +203,72 @@ class GeneratedServerTestSuite(CycladesTests):
"""Test server OS is actually up and running again"""
self.test_009_server_ping_ipv4()
def test_016_ssh_to_server_ipv4(self):
"""Test SSH to server public IPv4 works, verify hostname"""
self._skip_if(not self._image_is(self.use_image, "linux"),
"only valid for Linux servers")
hostname = self._insist_get_hostname_over_ssh(
self.ipv4, self.username, self.password)
# The hostname must be of the form 'prefix-id'
self.assertTrue(hostname.endswith("-%d" % self.server['id']))
def test_017_ssh_to_server_ipv6(self):
"""Test SSH to server public IPv6 works, verify hostname"""
self._skip_if(not self._image_is(self.use_image, "linux"),
"only valid for Linux servers")
self._skip_if(not self.use_ipv6, "--no-ipv6 flag enabled")
hostname = self._insist_get_hostname_over_ssh(
self.ipv6, self.username, self.password)
# The hostname must be of the form 'prefix-id'
self.assertTrue(hostname.endswith("-%d" % self.server['id']))
def test_018_rdp_to_server_ipv4(self):
"""Test RDP connection to server public IPv4 works"""
self._skip_if(not self._image_is(self.use_image, "windows"),
"only valid for Windows servers")
sock = self._insist_on_tcp_connection(socket.AF_INET, self.ipv4, 3389)
# No actual RDP processing done. We assume the RDP server is there
# if the connection to the RDP port is successful.
# pylint: disable-msg=W0511
# FIXME: Use rdesktop, analyze exit code? see manpage
sock.close()
def test_019_rdp_to_server_ipv6(self):
"""Test RDP connection to server public IPv6 works"""
self._skip_if(not self._image_is(self.use_image, "windows"),
"only valid for Windows servers")
self._skip_if(not self.use_ipv6, "--no-ipv6 flag enabled")
sock = self._insist_on_tcp_connection(socket.AF_INET, self.ipv6, 3389)
# No actual RDP processing done. We assume the RDP server is there
# if the connection to the RDP port is successful.
# pylint: disable-msg=W0511
# FIXME: Use rdesktop, analyze exit code? see manpage
sock.close()
def test_020_personality(self):
"""Test file injection for personality enforcement"""
self._skip_if(not self._image_is(self.use_image, "linux"),
"only implemented for linux servers")
assert self.personality is not None, "No personality used"
for inj_file in self.personality:
self._check_file_through_ssh(
self.ipv4, inj_file['owner'], self.password,
inj_file['path'], inj_file['contents'])
def test_021_submit_delete_request(self):
"""Test submit request to delete server"""
self.clients.cyclades.delete_server(self.server['id'])
def test_022_server_becomes_deleted(self):
"""Test server becomes DELETED"""
self._insist_on_server_transition(self.server, "ACTIVE", "DELETED")
def test_023_server_no_longer(self):
"""Test server is no longer in server list"""
servers = self._get_list_of_servers()
self.assertNotIn(self.server['id'], [s['id'] for s in servers])
# --------------------------------------------------------------------
# The actuall test class. We use this class to dynamically create
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment