OpenStack in a snap!
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

270 lines
8.6 KiB

import logging
import json
import unittest
import os
import subprocess
import time
import xvfbwrapper
from typing import List
import petname
from selenium import webdriver
from selenium.webdriver.common.by import By
# Setup logging
log = logging.getLogger("microstack_test")
log.setLevel(logging.DEBUG)
stream = logging.StreamHandler()
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s')
stream.setFormatter(formatter)
log.addHandler(stream)
def check(*args: List[str]) -> int:
"""Execute a shell command, raising an error on failed excution.
:param args: strings to be composed into the bash call.
"""
return subprocess.check_call(args)
def check_output(*args: List[str]) -> str:
"""Execute a shell command, returning the output of the command.
:param args: strings to be composed into the bash call.
Include our env; pass in any extra keyword args.
"""
return subprocess.check_output(args, universal_newlines=True).strip()
def call(*args: List[str]) -> bool:
"""Execute a shell command.
Return True if the call executed successfully (returned 0), or
False if it returned with an error code (return > 0)
:param args: strings to be composed into the bash call.
"""
return not subprocess.call(args)
def gui_wrapper(func):
"""Start up selenium drivers, run a test, then tear them down."""
def wrapper(cls, *args, **kwargs):
# Setup Selenium Driver
cls.display = xvfbwrapper.Xvfb(width=1280, height=720)
cls.display.start()
cls.driver = webdriver.PhantomJS()
# Run function
try:
return func(cls, *args, **kwargs)
finally:
# Tear down driver
cls.driver.quit()
cls.display.stop()
return wrapper
class Host():
"""A host with MicroStack installed."""
def __init__(self):
self.prefix = []
self.dump_dir = '/tmp'
self.machine = ''
self.distro = os.environ.get('DISTRO') or 'bionic'
self.snap = os.environ.get('SNAP_FILE') or \
'microstack_stein_amd64.snap'
self.horizon_ip = '10.20.20.1'
self.host_type = 'localhost'
if os.environ.get('MULTIPASS'):
self.host_type = 'multipass'
print("Booting a Multipass VM ...")
self.multipass()
def install(self, snap=None, channel='dangerous'):
if snap is None:
snap = self.snap
print("Installing {}".format(snap))
check(*self.prefix, 'sudo', 'snap', 'install', '--devmode',
'--{}'.format(channel), snap)
def init(self, flag='auto'):
print("Initializing the snap with --{}".format(flag))
check(*self.prefix, 'sudo', 'microstack.init', '--{}'.format(flag))
def multipass(self):
self.machine = petname.generate()
self.prefix = ['multipass', 'exec', self.machine, '--']
check('sudo', 'snap', 'install', '--classic', '--edge', 'multipass')
check('multipass', 'launch', '--cpus', '2', '--mem', '8G', self.distro,
'--name', self.machine)
check('multipass', 'copy-files', self.snap, '{}:'.format(self.machine))
# Figure out machine's ip
info = check_output('multipass', 'info', self.machine, '--format',
'json')
info = json.loads(info)
self.horizon_ip = info['info'][self.machine]['ipv4'][0]
def dump_logs(self):
# TODO: make unique log name
if check_output('whoami') == 'zuul':
self.dump_dir = "/home/zuul/zuul-output/logs"
check(*self.prefix,
'sudo', 'tar', 'cvzf',
'{}/dump.tar.gz'.format(self.dump_dir),
'/var/snap/microstack/common/log',
'/var/snap/microstack/common/etc',
'/var/log/syslog')
if 'multipass' in self.prefix:
check('multipass', 'copy-files',
'{}:/tmp/dump.tar.gz'.format(self.machine), '.')
print('Saved dump.tar.gz to local working dir.')
def teardown(self):
if 'multipass' in self.prefix:
check('sudo', 'multipass', 'delete', self.machine)
check('sudo', 'multipass', 'purge')
else:
if call('snap', 'list', 'microstack'):
# Uninstall microstack if it is installed (it may not be).
check('sudo', 'snap', 'remove', '--purge', 'microstack')
class Framework(unittest.TestCase):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.HOSTS = []
def get_host(self):
if self.HOSTS:
return self.HOSTS[0]
host = Host()
self.HOSTS.append(host)
return host
def add_host(self):
host = Host()
self.HOSTS.append(host)
return host
def verify_instance_networking(self, host, instance_name):
"""Verify that we have networking on an instance
We should be able to ping the instance.
And we should be able to reach the Internet.
"""
print("Skipping instance networking test due to bug #1852206")
# TODO re-enable this test when we have fixed
# https://bugs.launchpad.net/microstack/+bug/1852206
return True
prefix = host.prefix
# Ping the instance
print("Testing ping ...")
ip = None
servers = check_output(*prefix, '/snap/bin/microstack.openstack',
'server', 'list', '--format', 'json')
servers = json.loads(servers)
for server in servers:
if server['Name'] == instance_name:
ip = server['Networks'].split(",")[1].strip()
break
self.assertTrue(ip)
pings = 1
max_pings = 600 # ~10 minutes!
while not call(*prefix, 'ping', '-c1', '-w1', ip):
pings += 1
if pings > max_pings:
self.assertFalse(True, msg='Max pings reached!')
print("Testing instances' ability to connect to the Internet")
# Test Internet connectivity
attempts = 1
max_attempts = 300 # ~10 minutes!
username = check_output(*prefix, 'whoami')
while not call(
*prefix,
'ssh',
'-oStrictHostKeyChecking=no',
'-i', '/home/{}/.ssh/id_microstack'.format(username),
'cirros@{}'.format(ip),
'--', 'ping', '-c1', '91.189.94.250'):
attempts += 1
if attempts > max_attempts:
self.assertFalse(
True,
msg='Unable to access the Internet!')
time.sleep(1)
@gui_wrapper
def verify_gui(self, host):
"""Verify that Horizon Dashboard works
We should be able to reach the dashboard.
We should be able to login.
"""
# Test
print('Verifying GUI for (IP: {})'.format(host.horizon_ip))
# Verify that our GUI is working properly
dashboard_port = check_output(*host.prefix, 'sudo', 'snap', 'get',
'microstack',
'config.network.ports.dashboard')
self.driver.get("http://{}:{}/".format(
host.horizon_ip,
dashboard_port
))
# Login to horizon!
self.driver.find_element(By.ID, "id_username").click()
self.driver.find_element(By.ID, "id_username").send_keys("admin")
self.driver.find_element(By.ID, "id_password").send_keys("keystone")
self.driver.find_element(By.CSS_SELECTOR, "#loginBtn > span").click()
# Verify that we can click something on the dashboard -- e.g.,
# we're still not sitting at the login screen.
self.driver.find_element(By.LINK_TEXT, "Images").click()
def setUp(self):
self.passed = False # HACK: trigger (or skip) log dumps.
def tearDown(self):
"""Clean hosts up, possibly leaving debug information behind."""
print("Tests complete. Cleaning up.")
while self.HOSTS:
host = self.HOSTS.pop()
if not self.passed:
print(
"Tests failed. Leaving {} in place.".format(host.machine))
# Skipping log dump, due to
# https://bugs.launchpad.net/microstack/+bug/1860783
# host.dump_logs()
if os.environ.get('INTERACTIVE_DEBUG'):
print('INTERACTIVE_DEBUG set. '
'Opening a shell on test machine.')
call('multipass', 'shell', host.machine)
else:
print("Tests complete. Cleaning up.")
host.teardown()