tests: group and parametrize tests

Instead of creating a dedicated test and using the same testinfra
module we can group them into a single test to avoid multiple ansible
connections and testinfra module execution.
This patch also adds parametrize pytest decorator when possible.
Finally fixing some flake minor issue.

Signed-off-by: Dimitri Savineau <dsavinea@redhat.com>
pull/3924/head
Dimitri Savineau 2019-04-18 17:08:13 -04:00 committed by Guillaume Abrioux
parent 8ab6a3391f
commit 564ec9c992
8 changed files with 63 additions and 126 deletions

View File

@ -4,51 +4,21 @@ import pytest
class TestiSCSIs(object):
@pytest.mark.no_docker
def test_tcmu_runner_is_installed(self, node, host):
assert host.package("tcmu-runner").is_installed
@pytest.mark.parametrize('pkg', [
'ceph-iscsi-cli',
'ceph-iscsi-config',
'targetcli',
'tcmu-runner'
])
def test_iscsi_package_is_installed(self, node, host, pkg):
assert host.package(pkg).is_installed
@pytest.mark.no_docker
def test_ceph_iscsi_config_is_installed(self, node, host):
assert host.package("ceph-iscsi-config").is_installed
@pytest.mark.no_docker
def test_targetcli_is_installed(self, node, host):
assert host.package("targetcli").is_installed
# @pytest.mark.no_docker
# def test_rbd_target_api_is_installed(self, node, host):
# assert host.package("rbd-target-api").is_installed
# @pytest.mark.no_docker
# def test_rbd_target_gw_is_installed(self, node, host):
# assert host.package("rbd-target-gw").is_installed
@pytest.mark.docker
def test_tcmu_runner_service_is_running(self, node, host):
service_name = "tcmu-runner"
assert host.service(service_name).is_running
@pytest.mark.docker
def test_rbd_target_api_service_is_running(self, node, host):
service_name = "rbd-target-api"
assert host.service(service_name).is_running
@pytest.mark.docker
def test_rbd_target_gw_service_is_running(self, node, host):
service_name = "rbd-target-gw"
assert host.service(service_name).is_running
@pytest.mark.docker
def test_tcmu_runner_service_is_enabled(self, node, host):
service_name = "tcmu-runner"
assert host.service(service_name).is_enabled
@pytest.mark.docker
def test_rbd_target_api_service_is_enabled(self, node, host):
service_name = "rbd-target-api"
assert host.service(service_name).is_enabled
@pytest.mark.docker
def test_rbd_target_gw_service_is_enabled(self, node, host):
service_name = "rbd-target-gw"
assert host.service(service_name).is_enabled
@pytest.mark.parametrize('svc', [
'rbd-target-api',
'rbd-target-gw',
'tcmu-runner'
])
def test_iscsi_service_enabled_and_running(self, node, host, svc):
s = host.service(svc)
assert s.is_enabled
assert s.is_running

View File

@ -8,17 +8,13 @@ class TestMDSs(object):
def test_mds_is_installed(self, node, host):
assert host.package("ceph-mds").is_installed
def test_mds_service_is_running(self, node, host):
def test_mds_service_enabled_and_running(self, node, host):
service_name = "ceph-mds@{hostname}".format(
hostname=node["vars"]["inventory_hostname"]
)
assert host.service(service_name).is_running
def test_mds_service_is_enabled(self, node, host):
service_name = "ceph-mds@{hostname}".format(
hostname=node["vars"]["inventory_hostname"]
)
assert host.service(service_name).is_enabled
s = host.service(service_name)
assert s.is_enabled
assert s.is_running
def test_mds_is_up(self, node, host, setup):
hostname = node["vars"]["inventory_hostname"]

View File

@ -15,17 +15,13 @@ class TestMons(object):
port=mon_port
)).is_listening
def test_mon_service_is_running(self, node, host):
def test_mon_service_enabled_and_running(self, node, host):
service_name = "ceph-mon@{hostname}".format(
hostname=node["vars"]["inventory_hostname"]
)
assert host.service(service_name).is_running
def test_mon_service_is_enabled(self, node, host):
service_name = "ceph-mon@{hostname}".format(
hostname=node["vars"]["inventory_hostname"]
)
assert host.service(service_name).is_enabled
s = host.service(service_name)
assert s.is_enabled
assert s.is_running
@pytest.mark.no_docker
def test_can_get_cluster_health(self, node, host, setup):
@ -36,7 +32,7 @@ class TestMons(object):
def test_ceph_config_has_inital_members_line(self, node, File, setup):
assert File(setup["conf_path"]).contains("^mon initial members = .*$")
def test_initial_members_line_has_correct_value(self, node, host, File, setup):
def test_initial_members_line_has_correct_value(self, node, host, File, setup): # noqa E501
mon_initial_members_line = host.check_output("grep 'mon initial members = ' /etc/ceph/{cluster}.conf".format(cluster=setup['cluster_name'])) # noqa E501
result = True
for host in node["vars"]["groups"]["mons"]:

View File

@ -5,20 +5,18 @@ import pytest
class TestNFSs(object):
@pytest.mark.no_docker
def test_nfs_ganesha_is_installed(self, node, host):
assert host.package("nfs-ganesha").is_installed
@pytest.mark.parametrize('pkg', [
'nfs-ganesha',
'nfs-ganesha-rgw'
])
def test_nfs_ganesha_package_is_installed(self, node, host, pkg):
assert host.package(pkg).is_installed
@pytest.mark.no_docker
def test_nfs_ganesha_rgw_package_is_installed(self, node, host):
assert host.package("nfs-ganesha-rgw").is_installed
@pytest.mark.no_docker
def test_nfs_services_are_running(self, node, host):
assert host.service("nfs-ganesha").is_running
@pytest.mark.no_docker
def test_nfs_services_are_enabled(self, node, host):
assert host.service("nfs-ganesha").is_enabled
def test_nfs_service_enabled_and_running(self, node, host):
s = host.service("nfs-ganesha")
assert s.is_enabled
assert s.is_running
@pytest.mark.no_docker
def test_nfs_config_override(self, node, host):

View File

@ -1,6 +1,5 @@
import pytest
import json
import os
class TestOSDs(object):
@ -21,15 +20,12 @@ class TestOSDs(object):
assert host.check_output("netstat -lntp | grep ceph-osd | grep %s | wc -l" % # noqa E501
(setup["cluster_address"])) == str(nb_port)
def test_osd_services_are_running(self, node, host, setup):
def test_osd_service_enabled_and_running(self, node, host, setup):
# TODO: figure out way to paramaterize node['osds'] for this test
for osd in setup["osds"]:
assert host.service("ceph-osd@%s" % osd).is_running
def test_osd_services_are_enabled(self, node, host, setup):
# TODO: figure out way to paramaterize node['osds'] for this test
for osd in setup["osds"]:
assert host.service("ceph-osd@%s" % osd).is_enabled
s = host.service("ceph-osd@%s" % osd)
assert s.is_enabled
assert s.is_running
@pytest.mark.no_docker
def test_osd_are_mounted(self, node, host, setup):
@ -42,12 +38,12 @@ class TestOSDs(object):
assert host.mount_point(osd_path).exists
@pytest.mark.no_docker
def test_ceph_volume_is_installed(self, node, host):
assert host.exists('ceph-volume')
@pytest.mark.no_docker
def test_ceph_volume_systemd_is_installed(self, node, host):
assert host.exists('ceph-volume-systemd')
@pytest.mark.parametrize('cmd', [
'ceph-volume',
'ceph-volume-systemd'
])
def test_ceph_volume_command_exists(self, node, host, cmd):
assert host.exists(cmd)
def _get_osd_id_from_host(self, node, osd_tree):
children = []

View File

@ -8,24 +8,13 @@ class TestRbdMirrors(object):
def test_rbd_mirror_is_installed(self, node, host):
assert host.package("rbd-mirror").is_installed
@pytest.mark.docker
def test_rbd_mirror_service_is_running_docker(self, node, host):
def test_rbd_mirror_service_enabled_and_running(self, node, host):
service_name = "ceph-rbd-mirror@rbd-mirror.{hostname}".format(
hostname=node["vars"]["inventory_hostname"]
)
assert host.service(service_name).is_running
def test_rbd_mirror_service_is_running(self, node, host):
service_name = "ceph-rbd-mirror@rbd-mirror.{hostname}".format(
hostname=node["vars"]["inventory_hostname"]
)
assert host.service(service_name).is_running
def test_rbd_mirror_service_is_enabled(self, node, host):
service_name = "ceph-rbd-mirror@rbd-mirror.{hostname}".format(
hostname=node["vars"]["inventory_hostname"]
)
assert host.service(service_name).is_enabled
s = host.service(service_name)
assert s.is_enabled
assert s.is_running
def test_rbd_mirror_is_up(self, node, host, setup):
hostname = node["vars"]["inventory_hostname"]

View File

@ -11,21 +11,15 @@ class TestRGWs(object):
result = host.package("ceph-radosgw").is_installed
assert result
def test_rgw_service_is_running(self, node, host):
def test_rgw_service_enabled_and_running(self, node, host):
for i in range(int(node["radosgw_num_instances"])):
service_name = "ceph-radosgw@rgw.{hostname}.rgw{seq}".format(
hostname=node["vars"]["inventory_hostname"],
seq=i
)
assert host.service(service_name).is_running
def test_rgw_service_is_enabled(self, node, host):
for i in range(int(node["radosgw_num_instances"])):
service_name = "ceph-radosgw@rgw.{hostname}.rgw{seq}".format(
hostname=node["vars"]["inventory_hostname"],
seq=i
)
assert host.service(service_name).is_enabled
s = host.service(service_name)
assert s.is_enabled
assert s.is_running
def test_rgw_is_up(self, node, host, setup):
hostname = node["vars"]["inventory_hostname"]

View File

@ -4,17 +4,15 @@ import re
class TestInstall(object):
def test_ceph_dir_exists(self, host, node):
assert host.file('/etc/ceph').exists
def test_ceph_dir_exists_and_is_directory(self, host, node):
f = host.file('/etc/ceph')
assert f.exists
assert f.is_directory
def test_ceph_dir_is_a_directory(self, host, node):
assert host.file('/etc/ceph').is_directory
def test_ceph_conf_exists(self, host, node, setup):
assert host.file(setup["conf_path"]).exists
def test_ceph_conf_is_a_file(self, host, node, setup):
assert host.file(setup["conf_path"]).is_file
def test_ceph_conf_exists_and_is_file(self, host, node, setup):
f = host.file(setup["conf_path"])
assert f.exists
assert f.is_file
@pytest.mark.no_docker
def test_ceph_command_exists(self, host, node):
@ -27,7 +25,7 @@ class TestCephConf(object):
mon_host_line = host.check_output("grep 'mon host = ' /etc/ceph/{cluster}.conf".format(cluster=setup['cluster_name'])) # noqa E501
result = True
for x in range(0, setup["num_mons"]):
pattern = re.compile(("v2:{subnet}.1{x}:3300,v1:{subnet}.1{x}:6789".format(subnet=setup["subnet"], x=x)))
pattern = re.compile(("v2:{subnet}.1{x}:3300,v1:{subnet}.1{x}:6789".format(subnet=setup["subnet"], x=x))) # noqa E501
if pattern.search(mon_host_line) is None:
result = False
assert result