microstack/tests/framework.py

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 = 'bionic'
        self.snap = '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', '--classic',
              '--{}'.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, '--']
        distro = os.environ.get('distro') or self.distro

        check('sudo', 'snap', 'install', '--classic', '--edge', 'multipass')

        check('multipass', 'launch', '--cpus', '2', '--mem', '8G', 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()