#!/usr/bin/env python3
# Kayobe overcloud host configure tests.
# Uses py.test and TestInfra.
import ipaddress
import os
import distro
import pytest
def _is_dnf():
info = distro.linux_distribution()
return info[0] == 'CentOS Linux' and info[1].startswith('8')
def test_network_ethernet(host):
interface = host.interface('dummy2')
assert interface.exists
assert '192.168.34.1' in interface.addresses
routes = host.check_output('/sbin/ip route show dev dummy2')
assert '192.168.40.0/24 via 192.168.34.254' in routes
def test_network_ethernet_vlan(host):
interface = host.interface('dummy2.42')
assert interface.exists
assert '192.168.35.1' in interface.addresses
assert host.file('/sys/class/net/dummy2.42/lower_dummy2').exists
routes = host.check_output(
'/sbin/ip route show dev dummy2.42 table kayobe-test-route-table')
assert '192.168.40.0/24 via 192.168.35.254' in routes
rules = host.check_output(
'/sbin/ip rule show table kayobe-test-route-table')
expected = 'from 192.168.35.0/24 lookup kayobe-test-route-table'
assert expected in rules
def test_network_bridge(host):
interface = host.interface('br0')
assert interface.exists
assert '192.168.36.1' in interface.addresses
ports = ['dummy3', 'dummy4']
sys_ports = host.check_output('ls -1 /sys/class/net/br0/brif')
assert sys_ports == "\n".join(ports)
for port in ports:
interface = host.interface(port)
assert interface.exists
v4_addresses = [a for a in interface.addresses
if ipaddress.ip_address(a).version == '4']
assert not v4_addresses
def test_network_bridge_vlan(host):
interface = host.interface('br0.43')
assert interface.exists
assert '192.168.37.1' in interface.addresses
assert host.file('/sys/class/net/br0.43/lower_br0').exists
def test_network_bond(host):
interface = host.interface('bond0')
assert interface.exists
assert '192.168.38.1' in interface.addresses
sys_slaves = host.check_output('cat /sys/class/net/bond0/bonding/slaves')
# Ordering is not guaranteed, so compare sets.
sys_slaves = set(sys_slaves.split())
slaves = set(['dummy5', 'dummy6'])
assert sys_slaves == slaves
for slave in slaves:
interface = host.interface(slave)
assert interface.exists
assert not interface.addresses
def test_network_bond_vlan(host):
interface = host.interface('bond0.44')
assert interface.exists
assert '192.168.39.1' in interface.addresses
assert host.file('/sys/class/net/bond0.44/lower_bond0').exists
def test_additional_user_account(host):
user = host.user("kayobe-test-user")
assert user.name == "kayobe-test-user"
assert user.group == "kayobe-test-user"
assert set(user.groups) == {"kayobe-test-user", "stack"}
assert user.gecos == "Kayobe test user"
with host.sudo():
assert user.password == 'kayobe-test-user-password'
def test_software_RAID(host):
slaves = host.check_output("ls -1 /sys/class/block/md0/slaves/")
assert slaves == "loop0\nloop1"
def test_luks(host):
# blkid returns an emptry string without root permissions
with host.sudo():
blkid = host.check_output('blkid /dev/md0')
assert 'TYPE="crypto_LUKS"' in blkid
def test_sysctls(host):
assert host.sysctl("fs.mount-max") == 99999
def test_cloud_init_is_disabled(host):
assert host.file("/etc/cloud/cloud-init.disabled").exists
def test_docker_storage_driver_is_devicemapper(host):
with host.sudo("stack"):
info = host.check_output("docker info")
assert "devicemapper" in info
@pytest.mark.parametrize('user', ['kolla', 'stack'])
def test_docker_image_download(host, user):
with host.sudo(user):
host.check_output("docker pull alpine")
@pytest.mark.parametrize('user', ['kolla', 'stack'])
def test_docker_container_run(host, user):
with host.sudo(user):
host.check_output("docker run --rm alpine /bin/true")
def test_timezone(host):
status = host.check_output("timedatectl status")
assert "Pacific/Honolulu" in status
def test_ntp_alternative_services_disabled(host):
# Tests that we don't have any conflicting NTP servers running
# NOTE(wszumski): We always mask services even if they don't exist
ntpd_service = host.service("ntp")
assert ntpd_service.is_masked
assert not ntpd_service.is_running
timesyncd_service = host.service("systemd-timesyncd")
assert timesyncd_service.is_masked
assert not timesyncd_service.is_running
def test_ntp_running(host):
# Tests that NTP services are enabled and running
assert host.package("chrony").is_installed
assert host.service("chronyd").is_enabled
assert host.service("chronyd").is_running
def test_ntp_non_default_time_server(host):
# Tests that the NTP pool has been changed from pool.ntp.org to
# time.cloudflare.com
if 'centos' in host.system_info.distribution.lower():
chrony_config = host.file("/etc/chrony.conf")
else:
# Debian based distributions use the following path
chrony_config = host.file("/etc/chrony/chrony.conf")
assert chrony_config.exists
assert "time.cloudflare.com" in chrony_config.content_string
def test_ntp_clock_synchronized(host):
# Tests that the clock is synchronized
status_output = host.check_output("timedatectl status")
assert "synchronized: yes" in status_output
@pytest.mark.parametrize('repo', ["appstream", "baseos", "extras", "epel",
"epel-modular"])
@pytest.mark.skipif(not _is_dnf(), reason="DNF only supported on CentOS 8")
def test_dnf_local_package_mirrors(host, repo):
# Depends on SITE_MIRROR_FQDN environment variable.
assert os.getenv('SITE_MIRROR_FQDN')
# NOTE(mgoddard): Should not require sudo but some files
# (/var/cache/dnf/expired_repos.json) can have incorrect permissions.
# https://bugzilla.redhat.com/show_bug.cgi?id=1636909
with host.sudo():
info = host.check_output("dnf repoinfo %s", repo)
assert os.getenv('SITE_MIRROR_FQDN') in info
@pytest.mark.skipif(not _is_dnf(), reason="DNF only supported on CentOS 8")
def test_dnf_custom_package_repository_is_available(host):
with host.sudo():
host.check_output("dnf -y install td-agent")
assert host.package("td-agent").is_installed
@pytest.mark.skipif(not _is_dnf(), reason="DNF only supported on CentOS 8")
def test_dnf_automatic(host):
assert host.package("dnf-automatic").is_installed
assert host.service("dnf-automatic.timer").is_enabled
assert host.service("dnf-automatic.timer").is_running