microstack/tests/framework.py
Dmitrii Shcherbakov 35ad8fbef0 Use microstack-support interface, drop devmode
With https://github.com/snapcore/snapd/pull/8926 merged and snapd 2.53.2
released we can now test changes without devmode.

Change-Id: I9980b171e537530d67f0a7eed332147f06fe7c3c
2021-11-26 18:23:03 +03:00

456 lines
16 KiB
Python

import logging
import json
import subprocess
import unittest
import yaml
import petname
import tenacity
from selenium import webdriver
from selenium.webdriver.firefox.options import Options as FirefoxOptions
from selenium.webdriver.common.by import By
# Setup logging
logger = logging.getLogger("microstack_test")
logger.setLevel(logging.DEBUG)
stream = logging.StreamHandler()
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s')
stream.setFormatter(formatter)
logger.addHandler(stream)
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 TestHost:
def __init__(self):
pass
def destroy(self):
raise NotImplementedError
def check_output(self, args, **kwargs):
raise NotImplementedError
def call(self, args, **kwargs):
raise NotImplementedError
def check_call(self, args, **kwargs):
raise NotImplementedError
def install_snap(self, name, options):
self.check_output(['sudo', 'snap', 'install', name, *options])
def refresh_snap(self, name, options):
self.check_output(['sudo', 'snap', 'refresh', name, *options])
def try_snap(self, name):
try:
self.check_output(['unsquashfs', name])
except subprocess.CalledProcessError:
logger.warning("Re-using existing squashfs-root directory with "
"'snap try squashfs-root'")
self.check_output(['sudo', 'snap', 'try', 'squashfs-root'])
def remove_snap(self, name, options):
self.check_output(['sudo', 'snap', 'remove', name, *options])
def snap_connect(self, snap_name, plug_name):
self.check_output(['sudo', 'snap', 'connect',
f'{snap_name}:{plug_name}'])
def install_snapd(self, *, channel='stable'):
self.install_snap('snapd', [f'--{channel}'])
# In case snapd is already installed but with a different channel,
# refresh it to the right channel.
self.refresh_snap('snapd', [f'--{channel}'])
def install_microstack(self, *, channel='edge', path=None, snap_try=False):
"""Install MicroStack at this host and connect relevant plugs.
"""
if path and snap_try:
self.try_snap(path)
else:
if path is not None:
self.install_snap(path, ['--dangerous'])
else:
self.install_snap('microstack', [f'--{channel}'])
plugs = [
'libvirt', 'netlink-audit',
'firewall-control', 'hardware-observe',
'kernel-module-observe', 'kvm',
'log-observe', 'mount-observe',
'netlink-connector', 'network-observe',
'openvswitch-support', 'process-control',
'system-observe', 'network-control',
'system-trace', 'block-devices',
'raw-usb', 'microstack-support',
'hugepages-control',
]
for plug in plugs:
self.snap_connect('microstack', plug)
def init_microstack(self, args=['--auto']):
self.check_call(['sudo', 'microstack', 'init', *args])
def setup_tempest_verifier(self):
self.check_call(['sudo', 'snap', 'install', 'microstack-test'])
self.check_call(['sudo', 'mkdir', '-p',
'/tmp/snap.microstack-test/tmp'])
self.check_call(['sudo', 'cp',
'/var/snap/microstack/common/etc/microstack.json',
'/tmp/snap.microstack-test/tmp/microstack.json'])
self.check_call(['microstack-test.rally', 'db', 'recreate'])
self.check_call([
'microstack-test.rally', 'deployment', 'create',
'--filename', '/tmp/microstack.json',
'--name', 'snap_generated'])
self.check_call(['microstack-test.tempest-init'])
def _create_filtered_test_list(self, source):
target = '/tmp/snap.microstack-test/tmp/exclude-volume-tests.txt'
cmd = ['sudo', 'sh', '-c',
f'sed "/.*volume.*/d" {source} > {target}']
self.check_call(cmd)
return '/tmp/exclude-volume-tests.txt'
def run_verifications(self, include_volumes=False):
"""Run a set of verification tests on MicroStack from this host."""
test_list = '/snap/microstack-test/current/2020.06-test-list.txt'
if not include_volumes:
test_list = self._create_filtered_test_list(test_list)
self.check_call([
'microstack-test.rally', 'verify', 'start',
'--load-list', f'{test_list}',
'--detailed', '--concurrency', '2'])
self.check_call([
'microstack-test.rally', 'verify', 'report',
'--type', 'json', '--to',
'/tmp/verification-report.json'])
report = json.loads(self.check_output([
'sudo', 'cat',
'/tmp/snap.microstack-test/tmp/verification-report.json']))
# Make sure there are no verification failures in the report.
failures = list(report['verifications'].values())[0]['failures']
return failures
class LocalTestHost(TestHost):
def __init__(self):
super().__init__()
self.install_snap('multipass', ['--stable'])
self.install_snap('lxd', ['--stable'])
self.check_call(['sudo', 'lxd', 'init', '--auto'])
try:
self.run(['sudo', 'lxc', 'profile', 'show', 'microstack'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
check=True)
except subprocess.CalledProcessError as e:
stderr = e.stderr.decode('utf-8')
if 'No such object' in stderr:
self._create_microstack_profile()
return
else:
raise RuntimeError(
'An unexpected exception has occurred '
f'while trying to query the profile, stderr: {stderr}'
) from e
self.run(['sudo', 'lxc', 'profile', 'delete', 'microstack'],
check=True)
self._create_microstack_profile()
def _create_microstack_profile(self):
self.run(['sudo', 'lxc', 'profile', 'create', 'microstack'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True)
profile_conf = {
'config': {'linux.kernel_modules':
'iptable_nat, ip6table_nat, ebtables, openvswitch,'
'tap, vhost, vhost_net, vhost_scsi, vhost_vsock, kvm',
'security.nesting': 'true',
'limits.kernel.memlock': 'unlimited'
},
'devices':
{
'tun': {'path': '/dev/net/tun', 'type': 'unix-char'},
'vhost-net': {'path': '/dev/vhost-net', 'type': 'unix-char'},
'vhost-scsi': {'path': '/dev/vhost-scsi', 'type': 'unix-char'},
'vhost-vsock': {'path': '/dev/vhost-vsock',
'type': 'unix-char'},
'kvm': {'path': '/dev/kvm', 'type': 'unix-char'},
}
}
self.run(['sudo', 'lxc', 'profile', 'edit', 'microstack'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True,
input=yaml.dump(profile_conf).encode('utf-8'))
def destroy(self):
self.remove_snap('microstack', ['--purge'])
def check_output(self, args, **kwargs):
return subprocess.check_output(args, **kwargs).strip()
def call(self, args, **kwargs):
return subprocess.call(args, **kwargs)
def check_call(self, args, **kwargs):
subprocess.check_call(args, **kwargs)
def run(self, args, **kwargs):
subprocess.run(args, **kwargs)
class MultipassTestHost(TestHost):
"""A virtual host set up via Multipass on a local machine."""
def __init__(self, distribution):
self.distribution = distribution
self.name = petname.generate()
self._launch()
def check_output(self, args, **kwargs):
prefix = ['sudo', 'multipass', 'exec', self.name, '--']
cmd = []
cmd.extend(prefix)
cmd.extend(args)
return subprocess.check_output(cmd, **kwargs).strip()
def call(self, args, **kwargs):
prefix = ['sudo', 'multipass', 'exec', self.name, '--']
cmd = []
cmd.extend(prefix)
cmd.extend(args)
return subprocess.call(cmd, **kwargs)
def check_call(self, args, **kwargs):
prefix = ['sudo', 'multipass', 'exec', self.name, '--']
cmd = []
cmd.extend(prefix)
cmd.extend(args)
subprocess.check_call(cmd, **kwargs)
def run(self, args, **kwargs):
prefix = ['sudo', 'multipass', 'exec', self.name, '--']
cmd = []
cmd.extend(prefix)
cmd.extend(args)
subprocess.run(cmd, **kwargs)
def _launch(self):
# Possible upstream CI resource allocation is documented here:
# https://docs.opendev.org/opendev/infra-manual/latest/testing.html
# >= 8GiB of RAM
# 40-80 GB of storage (possibly under /opt)
# Swap is not guaranteed.
# With m1.tiny flavor the compute node needs slightly less than 3G of
# RAM and 2.5G of disk space.
subprocess.check_call(['sudo', 'sync'])
subprocess.check_call(['sudo', 'sh', '-c',
'echo 3 > /proc/sys/vm/drop_caches'])
subprocess.check_call(['sudo', 'multipass', 'launch', '--cpus', '2',
'--mem', '3G', '--disk', '4G',
self.distribution, '--name', self.name])
info = json.loads(subprocess.check_output(
['sudo', 'multipass', 'info', self.name,
'--format', 'json']))
self.address = info['info'][self.name]['ipv4'][0]
def copy_to(self, source_path, target_path=''):
"""Copy a file from the local machine to the Multipass VM.
"""
subprocess.check_call(['sudo', 'multipass', 'copy-files', source_path,
f'{self.name}:{target_path}'])
def copy_from(self, source_path, target_path):
"""Copy a file from the Multipass VM to the local machine.
"""
subprocess.check_call(['sudo', 'multipass', 'copy-files',
f'{self.name}:{source_path}',
target_path])
def destroy(self):
subprocess.check_call(['sudo', 'multipass', 'delete', self.name])
class LXDTestHost(TestHost):
"""A container test host set up via LXD on a local machine."""
def __init__(self, distribution):
self.distribution = distribution
self.name = petname.generate()
self._launch()
def check_output(self, args, **kwargs):
prefix = ['sudo', 'lxc', 'exec', self.name, '--']
cmd = []
cmd.extend(prefix)
cmd.extend(args)
return subprocess.check_output(cmd, **kwargs).strip()
def call(self, args, **kwargs):
prefix = ['sudo', 'lxc', 'exec', self.name, '--']
cmd = []
cmd.extend(prefix)
cmd.extend(args)
return subprocess.call(cmd, **kwargs)
def check_call(self, args, **kwargs):
prefix = ['sudo', 'lxc', 'exec', self.name, '--']
cmd = []
cmd.extend(prefix)
cmd.extend(args)
subprocess.check_call(cmd, **kwargs)
def run(self, args, **kwargs):
prefix = ['sudo', 'lxc', 'exec', self.name, '--']
cmd = []
cmd.extend(prefix)
cmd.extend(args)
subprocess.check_call(cmd, **kwargs)
def _launch(self):
subprocess.check_call(['sudo', 'lxc', 'launch',
f'ubuntu:{self.distribution}', self.name,
'--profile', 'default',
'--profile', 'microstack'])
@tenacity.retry(wait=tenacity.wait_fixed(3))
def fetch_addr_info():
info = json.loads(subprocess.check_output(
['sudo', 'lxc', 'query', f'/1.0/instances/{self.name}/state']))
addrs = info['network']['eth0']['addresses']
addr_info = next(filter(lambda a: a['family'] == 'inet', addrs),
None)
if addr_info is None:
raise RuntimeError('The container interface does'
' not have an IPv4 address which'
' is unexpected')
return addr_info
self.address = fetch_addr_info()['address']
def copy_to(self, source_path, target_path=''):
"""Copy file or directory to the container.
"""
subprocess.check_call(['sudo', 'lxc', 'file', 'push', source_path,
f'{self.name}/{target_path}',
'--recursive', '--create-dirs'])
def copy_from(self, source_path, target_path):
"""Copy file or directory from the container.
"""
subprocess.check_call(['sudo', 'lxc', 'file', 'pull'
f'{self.name}/{source_path}', target_path,
'--recursive', '--create-dirs'])
def destroy(self):
subprocess.check_call(['sudo', 'lxc', 'delete', self.name, '--force'])
class Framework(unittest.TestCase):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._test_hosts = []
def setUp(self):
self._localhost = LocalTestHost()
def tearDown(self):
for host in self._test_hosts:
host.destroy()
@property
def localhost(self):
return self._localhost
def add_multipass_host(self, distribution):
new_test_host = MultipassTestHost(distribution)
self._test_hosts.append(new_test_host)
return new_test_host
def add_lxd_host(self, distribution):
new_test_host = LXDTestHost(distribution)
self._test_hosts.append(new_test_host)
return new_test_host
def verify_instance_networking(self, test_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.
:param :class:`TestHost` test_host: The host to run the test from.
:param str instance_name: The name of the Nova instance to connect to.
"""
logger.debug("Testing ping ...")
ip = None
servers = test_host.check_output([
'/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)
test_host.call(['ping', '-i1', '-c10', '-w11', ip])
@gui_wrapper
def verify_gui(self, test_host):
"""Verify Horizon Dashboard operation by logging in."""
control_ip = test_host.check_output([
'sudo', 'snap', 'get', 'microstack', 'config.network.control-ip',
]).decode('utf-8')
logger.debug('Verifying GUI for (IP: {})'.format(control_ip))
dashboard_port = test_host.check_output([
'sudo', 'snap', 'get',
'microstack',
'config.network.ports.dashboard']).decode('utf-8')
keystone_password = test_host.check_output([
'sudo', 'snap', 'get',
'microstack',
'config.credentials.keystone-password'
]).decode('utf-8')
self.driver.get(f'https://{control_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_password)
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()