import logging import json import unittest import os import subprocess import time from typing import List import petname from selenium import webdriver from selenium.webdriver.firefox.options import Options as FirefoxOptions 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 options = FirefoxOptions() options.add_argument("-headless") cls.driver = webdriver.Firefox(options=options) # Run function try: return func(cls, *args, **kwargs) finally: # Tear down driver cls.driver.quit() 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()